D1: the assigner, builder and indexer call every node's manifest-reads again at their consumer's bind (issue 1966, ADR 0668 §4)

Note 155: a manifest-reads-changed said before the assigner's consumer existed was never heard, and
ace's bootstrapped statement stood. Each part registers the engine-reads keeper's AtBind on its hub
and calls the verb as a kept call (CallKeptOn). The asks-at-bind half (Q7) is not in this commit.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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

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

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
 }
 

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
 	}

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

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)

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

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