novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'The indexer publishes manifest-accepted and manifest-refused as events, without waiting for the stream's acknowledgement (issue 1966)' (#168) from fix/1966-indexer-event-refused into main

Browse Source 
Reviewed-on: #168
This commit was merged in pull request #168.
This commit is contained in:
jschoubben
2026-11-07 19:58:28 +00:00
parent 5013bb06e0 71a56d4606
commit a37ee99b41
2 changed files with 12 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+1 -1
View File
@@ -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),
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go
+11 -9
View File
@@ -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)
}
}