jschoubben
c623e8d235
A build ask its seat's stream gave up on, dropped or ended with no step was waited for with no end, and on restart taken over and waited for again: no build node would ever answer it, and nothing could ask it again (trunk key of mesh-build-node at 4e7b118e, building since 19:06; seven keys at 18:44). At the step bound, a context's bound and no timer of its own, the builder reads what the bus says of an ask with no step: gone from its stream, or first in line while a node took a later ask. Such an ask is asked again under a new id the key names, logged and said as build.asked-again, at most three askings, then the key fails with why. The bus's max-deliveries advisory would tell it sooner, but the builder is granted neither it nor the workers' information, and the controller is not changed here. builder_stuck reads every waiting key with its asks, askings and steps; builder_ask_again writes the new reasks state, which the builder watches and answers.
76 lines
2.7 KiB
Go
76 lines
2.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
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 credential, e build.AskedAgain) error {
|
|
data, err := json.Marshal(e)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + 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
|
|
}
|