From 71a56d4606c94985a7bb66157df443a44080a508 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 7 Nov 2026 20:57:02 +0100 Subject: [PATCH] The indexer publishes manifest-accepted and manifest-refused as events, without waiting for the stream's acknowledgement (issue 1966) Both were published with the JetStream acknowledgement awaited, a request and response, for ten seconds each, and manifest-accepted once more on failure. On a bus refusing the subject (the old grants do not grant the new indexer mesh.seat.mesh-catalog.event.manifest-accepted) no acknowledgement ever came, so each judging pass took about twenty seconds a manifest and the indexer never reached the forge's messages nor the seat definitions: the take-in stood still. Now both are published plainly on their subject with the dedup id in Nats-Msg-Id, as an event is (the operator, 2026-11-07: everything is event-driven); the stream that keeps the subject drops a duplicate by its id, and a refusal is said on the connection's error handler, as the journal already shows. An error of the publish itself (no connection) still goes to the caller, which retries once and writes the unsaid finding. Quick vs thorough: thorough for these two events (every publish of a seat event by the indexer is now an event); sayTakeIn keeps its acknowledged publish, said in the issue for 47/G. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/verdicts.go | 20 ++++++++++--------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index ca9980bb..ddfa67b1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -412,7 +412,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf) readWhenBound(seatReader, seatDefs, marks) ix := &index.Indexer{Findings: awaitedStore{findings}, OldSections: awaitedStore{oldSections}, Requirements: awaitedStore{requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, - Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred), + Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), ReadOpen: openVia(calls.Call), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go index e6ea8649..25a07db1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go @@ -15,7 +15,7 @@ import ( // sayRefused says the indexer's event manifest-refused (novox/hq ADR 0578 §1): mesh.seat.mesh-catalog.event.manifest-refused, // with the SDK's envelope headers and the digest and judge as the bus's dedup id, so one refusal is said once. -func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, ...any)) func(context.Context, meshcatalog.ManifestRefusedEvent) { +func sayRefused(conn *nats.Conn, cred process.Account, logf func(string, ...any)) func(context.Context, meshcatalog.ManifestRefusedEvent) { return func(ctx context.Context, r meshcatalog.ManifestRefusedEvent) { data, err := json.Marshal(r) if err != nil { @@ -29,10 +29,11 @@ func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, 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.Header.Set(nats.MsgIdHdr, id) msg.Data = data - pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) - defer cancel() - if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil { + // An event is published, never asked (issue 1966): the stream that keeps the subject takes it and drops a + // duplicate by its id; a bus refusing the subject says so on the connection's error handler, not by a wait here. + if err := conn.PublishMsg(msg); err != nil { logf("judge: %s", fmt.Errorf("saying %s for %s: %w", meshcatalog.EventManifestRefused, r.Module, err)) } } @@ -41,7 +42,7 @@ func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, // sayAccepted says the indexer's event manifest-accepted (issue 1966 D3, novox/hq ADR 0595 §4): // mesh.seat.mesh-catalog.event.manifest-accepted, carrying the module's bus-state entries whole, with the digest and judge as // the bus's dedup id (meshcatalog.ManifestAcceptedEvent.EventID), so one acceptance is said once. -func sayAccepted(js jetstream.JetStream, cred process.Account) func(context.Context, meshcatalog.ManifestAcceptedEvent, string) error { +func sayAccepted(conn *nats.Conn, cred process.Account) func(context.Context, meshcatalog.ManifestAcceptedEvent, string) error { return func(ctx context.Context, a meshcatalog.ManifestAcceptedEvent, id string) error { data, err := json.Marshal(a) if err != nil { @@ -53,11 +54,12 @@ func sayAccepted(js jetstream.JetStream, cred process.Account) func(context.Cont 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.Header.Set(nats.MsgIdHdr, id) msg.Data = data - pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) - defer cancel() - _, err = js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) - return err + // Published, never asked (issue 1966): the event's stream takes it and drops a duplicate by its id; waiting on + // the stream's acknowledgement held each judging pass twenty seconds a manifest on a bus refusing the subject. + // A refusal is said on the connection's error handler; an error here (no connection) is the caller's to retry. + return conn.PublishMsg(msg) } } -- 2.54.0