From 31a2bcdb3c76e05fdce38fccfc7a20fc6333fafd Mon Sep 17 00:00:00 2001
From: jochen <jschoubben@gmail.com>
Date: Sun, 25 Oct 2026 18:38:45 +0100
Subject: [PATCH] Read each resolved config key once when the consumer is made,
 so values set before this release reach the kept compositions (issue 1945,
 ADR 0545, review 3)

The config consumer delivered new writes only, and mesh-config never writes an
unchanged value again: g14's mesh-build-go capacity of 2, set before this
release, would never reach its kept compositions carrying 0. A newly made
durable now reads each key's last value once, through the same gather,
per-write lock and acknowledgement; its deliver policy is set before its first
delivery. Acknowledgements are synchronous and sent past the stop, and one the
bus refuses is said by its key.
Research 072: milestone 2, the composer reacting to the change of an input.
---
 cmd/mesh-controller/composed_start_test.go |  6 ++++++
 cmd/mesh-controller/kept_compositions.go   | 16 ++++++++++++----
 2 files changed, 18 insertions(+), 4 deletions(-)

diff --git a/cmd/mesh-controller/composed_start_test.go b/cmd/mesh-controller/composed_start_test.go
index 0f988d51..d3b4054b 100644
--- a/cmd/mesh-controller/composed_start_test.go
+++ b/cmd/mesh-controller/composed_start_test.go
@@ -131,6 +131,12 @@ func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) {
 				t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want)
 			}
 		}
+		// A value set before the consumer was made reaches the kept compositions: it reads each key once at creation.
+		if c := configFollowConfig(); c.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy ||
+			c.AckPolicy != jetstream.AckExplicitPolicy {
+			t.Errorf("the config consumer delivers %v, acknowledged %v; want each key's last value, acknowledged",
+				c.DeliverPolicy, c.AckPolicy)
+		}
 	})
 }
 
diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go
index 8258ff45..87aabfed 100644
--- a/cmd/mesh-controller/kept_compositions.go
+++ b/cmd/mesh-controller/kept_compositions.go
@@ -186,11 +186,15 @@ var configFollowGather = time.Second
 // configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3).
 const configFollowConsumer = broker.ConfigFollowConsumer
 
-// configFollowConfig is that consumer: each write from when it is made (a first start composes again nothing it did not
-// hear), acknowledged after the write is handled, so what a stop cut is delivered to the next controller.
+// configFollowConfig is that consumer: each resolved key's last value once when it is made, then each write,
+// acknowledged after the write is handled, so what a stop cut is delivered to the next controller. The first start of
+// the release that makes it composes again the kept compositions of every module with config, through the same gather,
+// lock and acknowledgement: mesh-config never writes an unchanged value again, so a value set before this release (g14's
+// mesh-build-go capacity of 2, against kept compositions carrying 0) would otherwise never reach them. Its deliver
+// policy is set before its first delivery; an existing durable's cannot be changed.
 func configFollowConfig() jetstream.ConsumerConfig {
 	return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>",
-		DeliverPolicy: jetstream.DeliverNewPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait,
+		DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait,
 		MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"}
 }
 
@@ -320,7 +324,11 @@ func followConfigOnce(ctx context.Context, open *stores) error {
 				if failed {
 					_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
 				} else {
-					_ = m.Ack()
+					acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+					if err := m.DoubleAck(acking); err != nil {
+						fmt.Printf("config: the bus did not take the acknowledgement of %s: %v\n", m.Subject(), err)
+					}
+					cancel()
 				}
 			}
 		}