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) } }