diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 9e72e33e..48ced1a4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -16,6 +16,7 @@ import ( "mesh-delivery-assigner/internal/buckets" "os" "os/signal" + "maps" "slices" "syscall" "time" @@ -315,7 +316,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, // from the node's record where the engine does not answer (the fallback, said in the journal). reads := &enginereads.Keeper{Fallback: w.recordReads, Changed: wake, Logf: logger.Printf} if bus != nil { - reads.Call = bus.CallOn + reads.Call = bus.CallKeptOn // a kept call, called again through a transport fault (ADR 0668 §1) } if kv := kvs[engineReadsBucket]; kv != nil { reads.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err } @@ -570,6 +571,9 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process. } } err := hub.Handle(enginereads.SubjectPrefix+"*", handle) + // At each bind every node the assigner holds a statement of is asked again: an event said before its consumer existed + // is not delivered to it (issue 1966 note 155: ace's bootstrapped statement stood; ADR 0668 §4). + hub.OnBound(reads.AtBind(func() []string { return slices.Sorted(maps.Keys(reads.Statements())) })) if err == nil { err = hub.Start(ctx, js, account) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 8d116ce6..ada4fe06 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -316,6 +316,9 @@ func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) { func (b *fakeBus) CallOn(node, key string, body any) (json.RawMessage, error) { return b.Call(node+"/"+key, body) } +func (b *fakeBus) CallKeptOn(node, key string, body any) (json.RawMessage, error) { + return b.CallOn(node, key, body) +} func (b *fakeBus) Emit(eventType string, body any) error { b.events = append(b.events, fmt.Sprintf("%s %+v", eventType, body)) return nil diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 492f4219..18540885 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -41,6 +41,8 @@ const PutEvent = "put" type caller interface { Call(key string, body any) (json.RawMessage, error) CallOn(node, key string, body any) (json.RawMessage, error) + // CallKeptOn is CallOn for a kept call, safe to call again (process.Conn.CallKeptOn; ADR 0668). + CallKeptOn(node, key string, body any) (json.RawMessage, error) Emit(eventType string, body any) error } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 224b428b..24f16218 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -263,7 +263,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro watch.put = gate.Toolchain // What each counted node's engine reads (ADR 0618 §2, ADR 0624 §1): asked through the node-engine seat's verb as the // builder's own account, for every node the assignments names, and kept in the state engine-reads once it is bound. - reads := newEngineReads(calls.CallOn, gate, logger.Printf) + reads := newEngineReads(calls.CallKeptOn, gate, logger.Printf) // The old reference form's window over them (DLV-B9), kept in the state old-form once it is bound; builder_old_form // reads it and drops a retired node by hand. reads.ctx = context.WithoutCancel(ctx) @@ -351,6 +351,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hub.Handle("mesh.seat.*.event.>", events); err != nil { return err } + hub.OnBound(reads.atBind) // what a node's engine said before the consumer was bound (ADR 0668 §4) if stopped != nil { return stopped } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go index 4475e766..b014f154 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go @@ -4,6 +4,7 @@ import ( "context" "slices" "sync" + "time" "github.com/nats-io/nats.go/jetstream" @@ -35,7 +36,7 @@ type engineReads struct { ctx context.Context } -// newEngineReads asks through call (process.Conn.CallOn) and keeps each answer in the state once it is bound. +// newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and keeps each answer in the state once it is bound. func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, logf func(string, ...any)) *engineReads { r := &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf}} r.keeper.Changed = r.stepWindow @@ -107,6 +108,16 @@ func (r *engineReads) placed() { r.stepWindow() } +// atBind asks every counted node again when the builder's consumer is bound (the hub's OnBound): a manifest-reads-changed +// said before the consumer existed is not delivered to it, so the statement it would have ended is ended here (novox/hq +// ADR 0668 §4; issue 1966 note 155). +func (r *engineReads) atBind(created time.Time) { + r.mu.Lock() + nodes := slices.Clone(r.counted) + r.mu.Unlock() + r.keeper.AtBind(func() []string { return nodes })(created) +} + // heard takes an event of the builder's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true. func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool { node, ours, err := enginereads.NodeOf(m.Subject(), m.Data()) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index f4d30f13..0994e9e9 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -414,6 +414,18 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { said: func(b versions.OldFormClosedBody) { closings = append(closings, b) }, log: func(e builds.BuildLogEntry) { logged = append(logged, e) }} reads := &engineReads{keeper: &enginereads.Keeper{}, window: window, counted: []string{"a", "b"}} + // Issue 1966 note 155, ADR 0668 §4: at the consumer's bind every counted node is called once again. + var boundCalls []string + reads.keeper.Sync = true + reads.keeper.Call = func(node, _ string, _ any) (json.RawMessage, error) { + boundCalls = append(boundCalls, node) + return nil, errors.New("not served") + } + reads.atBind(now) + if !slices.Equal(boundCalls, []string{"a", "b"}) { + t.Fatalf("at the bind called %v, want every counted node once", boundCalls) + } + reads.keeper = &enginereads.Keeper{} newest := []string{forms.FormNeed("oci")} counted := map[string][]string{"a": newest, "b": nil} window.step(context.Background(), []string{"a", "b"}, counted) 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 00ef9a3d..b2ae8c70 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -489,7 +489,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // What each counted node's engine reads (ADR 0624 §1), for the mesh-wide hold of a manifest every node reads (ADR 0632 // §2): asked through the node-engine seat's verb as the indexer's own account, for every node the assignments names, // kept in the state engine-reads once it is bound; each change weighs the holds again. - reads := newEngineReads(calls.CallOn, gate, func() { ix.KickHolds(ctx) }, logger.Printf) + reads := newEngineReads(calls.CallKeptOn, gate, func() { ix.KickHolds(ctx) }, logger.Printf) gate.OnPlaced = reads.placed ix.Counted, ix.AskEngines = reads.counted, reads.ask // The holds, new with ADR 0632 (issue 1790's kind), bound when they appear: until then a merge that would hold a @@ -648,6 +648,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }); err != nil { return err } + hub.OnBound(reads.atBind) // what a node's engine said before the consumer was bound (ADR 0668 §4) versions := func(m jetstream.Msg) { if !heardVersion(ctx, ix, m, logger.Printf) { _ = m.Ack() // nothing else the indexer consumes rides the shared stream diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go index 29a7c890..890a8f10 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go @@ -4,6 +4,7 @@ import ( "context" "slices" "sync" + "time" "github.com/nats-io/nats.go/jetstream" @@ -29,7 +30,7 @@ type engineReads struct { nodes []string } -// newEngineReads asks through call (process.Conn.CallOn) and tells changed after each statement changed. +// newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and tells changed after each statement changed. func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, changed func(), logf func(string, ...any)) *engineReads { return &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf, Changed: changed}} } @@ -77,6 +78,15 @@ func (r *engineReads) placed() { } } +// atBind asks every counted node again when the indexer's consumer is bound (the hub's OnBound): a manifest-reads-changed +// said before the consumer existed is not delivered to it (novox/hq ADR 0668 §4; issue 1966 note 155). +func (r *engineReads) atBind(created time.Time) { + r.mu.Lock() + nodes := slices.Clone(r.nodes) + r.mu.Unlock() + r.keeper.AtBind(func() []string { return nodes })(created) +} + // heard takes an event of the indexer's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true. func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool { node, ours, err := enginereads.NodeOf(m.Subject(), m.Data())