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