2026-10-21 09:05:56 +00:00
3 changed files with 136 additions and 12 deletions
@@ -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 {
@@ -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))
+	}
+}
@@ -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")
+	}
+}