mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"mesh-delivery-builder/internal/build"
)

// askAgainer is the part of the builder a reasks value asks through.
type askAgainer interface {
	AskAgain(key, why string) []string
	BuildAgain(ctx context.Context, module, fp, why, by string, bound time.Duration) build.BuildAgainResult
}

// buildAgainBound is how long the builder waits for the claim and the asks of a key built again by hand; the tool waits
// longer for the builder's answer.
const buildAgainBound = 15 * time.Second

// reAsked answers one value of the reasks state (the tool builder_ask_again, issue 1945): a value not yet answered asks the
// key's waiting lanes again and is written back, by compare-and-set at its revision, with what the builder did. A value
// answered already (this builder's own write) or deleted is left.
func reAsked(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) {
	if e.Operation() != jetstream.KeyValuePut {
		return
	}
	r, err := build.DecodeReAsk(e.Value())
	if err != nil {
		logf("%s.%s holds what is not an ask to ask again, left: %v", build.ReAsksState, e.Key(), err)
		return
	}
	if r.Answer != "" {
		return
	}
	if r.Mode == build.ModeBuild {
		// The answer comes when the claim and the asks are written, so it is made beside the watch, not in it.
		go builtAgain(ctx, b, kv, e, r, logf)
		return
	}
	asked := b.AskAgain(e.Key(), r.Why)
	if len(asked) == 0 {
		r.Answer = "no ask of this key waits in this builder now (it is not building, or its build has not reached its ask yet): nothing was asked; builder_stuck shows where it stands"
	} else {
		r.Answer = "asked again: the " + strings.Join(asked, ", ") + " ask, under a new id the key names"
	}
	r.Answered = time.Now().UTC()
	logf("%s: asked again by %s (%s): %s", e.Key(), r.By, r.Why, r.Answer)
	data, err := json.Marshal(r)
	if err != nil {
		return
	}
	if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
		logf("%s.%s: the answer could not be written: %v", build.ReAsksState, e.Key(), err)
	}
}

// emitAskedAgain says the builder's event build.asked-again (issue 1945): mesh.mod.<module>.event.build.asked-again,
// with the SDK's envelope headers and the new ask's id as the bus's dedup id.
func emitAskedAgain(ctx context.Context, js jetstream.JetStream, cred process.Account, e build.AskedAgain) error {
	data, err := json.Marshal(e)
	if err != nil {
		return err
	}
	msg := nats.NewMsg(cred.EventSubject(build.AskedAgainEvent))
	id := "asked-again-" + e.New
	msg.Header.Set("x-event-id", id)
	msg.Header.Set("x-source", cred.Module)
	msg.Header.Set("x-node", cred.Node)
	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
	msg.Header.Set("content-type", "application/json")
	msg.Data = data
	pub, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil {
		return fmt.Errorf("saying %s for %s: %w", build.AskedAgainEvent, e.Key, err)
	}
	return nil
}

// builtAgain answers a reasks value of mode build (the tool builder_build_again, issue 1126 note 72): the failed key is
// built again as a new revision, and the value is written back with the revision and the asks put, or why nothing was done.
func builtAgain(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstream.KeyValueEntry, r build.ReAsk, logf func(string, ...any)) {
	res := b.BuildAgain(ctx, r.Module, r.FP, r.Why, r.By, buildAgainBound)
	switch {
	case res.Refused != "":
		r.Answer, r.Refused = res.Refused, true
	case res.Failed != "":
		r.Answer, r.Revision, r.Failed = fmt.Sprintf("the key is written building as revision %d, and the build failed before any ask was put: %s", res.Revision, res.Failed), res.Revision, res.Failed
	default:
		r.Pending = res.Revision == 0
		r.Answer, r.Revision, r.Asks = "built again: the key is written building as revision "+fmt.Sprint(res.Revision)+" and its asks are put", res.Revision, res.Asks
		if res.Pending != "" {
			r.Answer = res.Pending
			if res.Revision != 0 {
				r.Answer = "built again: the key is written building as revision " + fmt.Sprint(res.Revision) + "; " + res.Pending
			}
		}
	}
	r.Answered = time.Now().UTC()
	logf("%s: build again asked by %s (%s): %s", e.Key(), r.By, r.Why, r.Answer)
	data, err := json.Marshal(r)
	if err != nil {
		return
	}
	if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
		logf("%s.%s: the answer could not be written: %v", build.ReAsksState, e.Key(), err)
	}
}