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

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