The indexer publishes manifest-accepted and manifest-refused as events, without waiting for the stream's acknowledgement (issue 1966)
mesh/delivery merged: source.main moved to novox/mesh-delivery@a37ee99b41d5
mesh/check pass 71a56d4606c9 on b73315ffe05c | carried from a37ee99b41d5 | unit 495 run, 947 kept | integration 494 run, 1014 kept | composition passe…

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
This commit is contained in:
2026-11-07 20:57:02 +01:00
parent 5013bb06e0
commit 71a56d4606
2 changed files with 12 additions and 10 deletions
@@ -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),
@@ -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)
}
}