diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index e1e609dc..9898ee8d 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "os" "slices" @@ -170,6 +171,21 @@ func (f mergeFacts) plans() bool { return len(f.from) > 0 || len(f.packaging) > 0 || len(f.later) > 0 || f.owed } +// heldMerge marks a failure to act on a merge, whatever it was, as "not now": the announcement is held on the bus and +// handed over again, so the merge joins the next batch, never taken (issue 1583, after 921). The bus acknowledges it +// only when hearMerge answers without ErrTryAgain, and a merge heard once is not announced again: the bus answering +// "jetstream not enabled", a bucket missing for a moment or a store read cut short each lost the merge when only an +// outage of a known kind (notNow) was held. The delay between hands-over is bounded (link.MergeRedeliverAfter). +func heldMerge(err error) error { + if err == nil { + return nil + } + if errors.Is(err, link.ErrTryAgain) { + return err + } + return fmt.Errorf("%w: the merge is not kept yet: %w", link.ErrTryAgain, err) +} + // hearMerge puts a merge the forge announced into the batch that answers it, at the moment it was heard, // and cuts what is due. func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.Time) error { @@ -180,7 +196,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T inv := f.open.inventory repository := m.Owner + "/" + m.Repo if kept, known, err := inv.MergeOf(ctx, repository, m.Commit); err != nil { - return notNow(err) + return heldMerge(err) } else if known { fmt.Printf("%s merged into %s (%.8s) was heard before; %s answers it\n", repository, m.Base, m.Commit, orWords(kept.Plan, "a walk of its own commit, once the walks before it end")) @@ -188,15 +204,15 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T } entries, err := inv.Catalogued(ctx) if err != nil { - return notNow(err) + return heldMerge(err) } read, err := readForPlanning(ctx, inv) if err != nil { - return notNow(err) + return heldMerge(err) } facts, err := factsOfMerge(ctx, inv, m, now, entries, read) if err != nil { - return notNow(err) + return heldMerge(err) } merged, touches := facts.merged, facts.touches if !facts.plans() { @@ -210,7 +226,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T } if already == 0 { if done, err := f.firstBuilds(ctx, m, entries, read, now); err != nil || done { - return err + return heldMerge(err) } } fmt.Print(nothingReadsIt(repository, m)) @@ -220,16 +236,16 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T own := ownPath(touches) waits, holding, err := groupWaitHolding(ctx, inv, m, deliverySeatHeld(entries)) if err != nil { - return notNow(err) + return heldMerge(err) } release, err := inv.HoldPlans(ctx, true) if err != nil { - return notNow(err) + return heldMerge(err) } defer release() // Again under the hold: a controller handing over may have kept it since. if _, known, err := inv.MergeOf(ctx, repository, m.Commit); err != nil || known { - return notNow(err) + return heldMerge(err) } var moves []string for _, e := range touches { @@ -248,11 +264,11 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T // word** (decided during the build, 2026-10-10): batched together, the catalogue's deliveries would start // with the controller's and skip their turn. Each kind has a batch of its own. if p, ok, err := answeredByALaterMerge(ctx, inv, heard, touches, own, deliverySeatHeld(entries)); err != nil { - return notNow(err) + return heldMerge(err) } else if ok { heard.Plan = p.ID if _, err := inv.AddMerge(ctx, heard); err != nil { - return notNow(err) + return heldMerge(err) } if p.Batch() { if err := keepBatch(ctx, inv, &p, now, ""); err != nil { @@ -268,11 +284,11 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own, waits) if err != nil { - return notNow(err) + return heldMerge(err) } heard.Plan = batch.ID if added, err := inv.AddMerge(ctx, heard); err != nil || !added { - return notNow(err) + return heldMerge(err) } holdFor(&batch, m.Head, waits, holding, own, now) if err := keepBatch(ctx, inv, &batch, now, ""); err != nil { diff --git a/cmd/mesh-controller/merge_bus_error_integration_test.go b/cmd/mesh-controller/merge_bus_error_integration_test.go new file mode 100644 index 00000000..81a56d04 --- /dev/null +++ b/cmd/mesh-controller/merge_bus_error_integration_test.go @@ -0,0 +1,75 @@ +//go:build integration + +package main + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/link" +) + +// withoutJetStream is a bucket whose reads fail the way a bus answers while its streams are away. +type withoutJetStream struct { + chain.Backend + away atomic.Bool +} + +func (w *withoutJetStream) Get(ctx context.Context, key string) (chain.Entry, bool, error) { + if w.away.Load() { + return chain.Entry{}, false, nats.ErrJetStreamNotEnabled + } + return w.Backend.Get(ctx, key) +} + +func (w *withoutJetStream) Keys(ctx context.Context) ([]string, error) { + if w.away.Load() { + return nil, nats.ErrJetStreamNotEnabled + } + return w.Backend.Keys(ctx) +} + +func (w *withoutJetStream) Watch(ctx context.Context) (<-chan chain.Entry, error) { + if w.away.Load() { + return nil, nats.ErrJetStreamNotEnabled + } + return w.Backend.Watch(ctx) +} + +func (w *withoutJetStream) WatchFiltered(ctx context.Context, filter string) (<-chan chain.Entry, error) { + if w.away.Load() { + return nil, nats.ErrJetStreamNotEnabled + } + return w.Backend.WatchFiltered(ctx, filter) +} + +// **A merge heard while the bus answers "jetstream not enabled" is held, recorded nowhere, and recorded when handed +// over again** (issue 1583, mesh-catalog #404): hearMerge answers ErrTryAgain, so the bus does not acknowledge the +// announcement, and the merge joins a batch when the bus is back. +func TestAMergeHeardWhileTheBusHasNoJetStreamIsHeldAndRecordedWhenHandedOverAgain(t *testing.T) { + open := windowed(t) + everyHeadWaits(t) + asksWithPaths(t) + ctx := t.Context() + assignments := &withoutJetStream{Backend: chain.NewMemory()} + open.inventory.UseChain(chain.New(chain.NewMemory(), chain.NewMemory(), assignments, func(context.Context) error { return nil })) + merge := catalogueMerge("cat2bbbbbbbb", "app", t0) + + assignments.away.Store(true) + err := (following{open: open}).hearMerge(ctx, merge, t0.Add(time.Second)) + if !errors.Is(err, link.ErrTryAgain) { + t.Fatalf("a merge heard while the bus had no jetstream answered %v, which the bus takes as done", err) + } + + assignments.away.Store(false) + hear(t, open, merge, t0.Add(10*time.Second)) + if _, bs := walks(t, open); len(bs) != 1 { + t.Fatalf("the merge handed over again made %d batch(es)", len(bs)) + } +} diff --git a/cmd/mesh-controller/merge_bus_error_test.go b/cmd/mesh-controller/merge_bus_error_test.go new file mode 100644 index 00000000..cf660e25 --- /dev/null +++ b/cmd/mesh-controller/merge_bus_error_test.go @@ -0,0 +1,33 @@ +package main + +import ( + "errors" + "testing" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/link" +) + +// **A merge the controller cannot act on is held, whatever the error** (issue 1583, after 921): the bus acknowledges an +// announcement only when hearMerge answers without ErrTryAgain, and a merge heard once is not announced again. The bus +// answering "jetstream not enabled" (mesh-catalog #404, 2026-10-21) is no outage of a kind notNow knows, and took the merge. +func TestAMergeThatFailsOnAnyErrorIsHeldAndNotTaken(t *testing.T) { + for name, cause := range map[string]error{ + "jetstream not enabled": nats.ErrJetStreamNotEnabled, + "a bucket missing": errors.New("the bucket mesh-controller_builds is not on the bus"), + "no responders": nats.ErrNoResponders, + } { + held := heldMerge(cause) + if !errors.Is(held, link.ErrTryAgain) || !errors.Is(held, cause) { + t.Errorf("%s answers %v, which the bus takes as done", name, held) + } + } + if heldMerge(nil) != nil { + t.Error("a merge that was heard answers an error") + } + already := link.ErrTryAgain + if heldMerge(already) != already { + t.Error("a merge already held is wrapped again") + } +}