amadan

carlos / platform Public

zdd-5 task 7: delivered and active config revisions; the delivery report

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 89008101bc30cb7d92b3bce2ced7575ca5223f1e parent ceafa25
5 files changed, +202 −1
  • internal/appconfig/claimsync.go +7 −1
  • internal/appconfig/syncexec.go +7 −0
  • internal/appconfig/syncexec_test.go +53 −0
  • internal/rollout/config.go +76 −0
  • internal/rollout/config_test.go +59 −0
diff --git a/internal/appconfig/claimsync.go b/internal/appconfig/claimsync.go
index 5e8bcf9af..44a7d87b9 100644
--- a/internal/appconfig/claimsync.go
+++ b/internal/appconfig/claimsync.go
@@ -48,7 +48,13 @@ type Callbacks struct {
Ack func(ConfigAck)
SelectedRestart hostwrite.Callback
SelectedAck func(context.Context, hostwrite.Selection, ConfigAck) error
- enabled bool
+ // Delivered, when set, is told the revision of a route's config file
+ // each time this pass writes or removes it, before the restart callback
+ // runs: the zero-downtime scheduler's "delivered" revision (spec §8.1).
+ // It is never called for a shape whose process the platform does not
+ // restart (the TCP-addr and sidecar shapes). Nil keeps today's behaviour.
+ Delivered func(host, rev string)
+ enabled bool
}
func (c Callbacks) restart(ctx context.Context, selected hostwrite.Selection) error {
diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go
index d1304717c..9f01a2d7f 100644
--- a/internal/appconfig/syncexec.go
+++ b/internal/appconfig/syncexec.go
@@ -616,6 +616,7 @@ func SyncExecWithCallbacks(ctx context.Context, reg *registry.Registry, s releas
if hostBounce == nil {
hostCallbacks.Restart = nil
hostCallbacks.SelectedRestart = nil
+ hostCallbacks.Delivered = nil
}
if err := syncExecHostClaimed(ctx, selected[rt.Host], rt, m, Key(k.acct, k.app), priv, dataDir, hostBounce, header, instETag, ack, hostCallbacks); err != nil && hostWrites.Enabled {
return err
@@ -1218,6 +1219,9 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R
if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
return err
}
+ if callbacks.Delivered != nil {
+ callbacks.Delivered(rt.Host, id.Config)
+ }
if bounce != nil || callbacks.SelectedRestart != nil {
if callbacks.enabled {
return callbacks.restart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id})
@@ -1237,6 +1241,9 @@ func syncExecHostClaimed(ctx context.Context, work hostwrite.Work, rt registry.R
if err := guarded(ctx, work, id, func() error { return nil }); err != nil {
return err
}
+ if callbacks.Delivered != nil {
+ callbacks.Delivered(rt.Host, id.Config)
+ }
var bounceErr error
if bounce != nil || callbacks.SelectedRestart != nil {
if callbacks.enabled {
diff --git a/internal/appconfig/syncexec_test.go b/internal/appconfig/syncexec_test.go
index 333df7ba5..8a46d6e9b 100644
--- a/internal/appconfig/syncexec_test.go
+++ b/internal/appconfig/syncexec_test.go
@@ -2480,3 +2480,56 @@ func TestSyncExecWritesConfigAckForTCPExecFileRoute(t *testing.T) {
t.Fatalf("ack.RelaySuffixes[eu-west-1] = %q, want %q", got.RelaySuffixes["eu-west-1"], wantSuffixes["eu-west-1"])
}
}
+
+// With overlap swaps armed, config sync reports each delivered revision to
+// the scheduler before its (no-op) restart, and never for a shape the
+// platform does not restart (zero-downtime spec §8.1).
+func TestSyncExecReportsTheDeliveredRevision(t *testing.T) {
+ ctx, s, reg, dd, priv, _ := execKit(t)
+ pk, err := ParsePrivateKey(priv, "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := SetEnv(ctx, s, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
+ t.Fatal(err)
+ }
+ for _, rt := range []registry.Route{
+ {Host: "inst.example.com", Kind: "instance", App: "eleven", Account: "bab", Status: "ready", Socket: "/run/carlos/inst.sock"},
+ {Host: "bot.eleven.bab.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bab", Status: "ready"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ var order, revs []string
+ cb := Callbacks{
+ Restart: func(h string) error { order = append(order, "restart:"+h); return nil },
+ Delivered: func(h, rev string) { order = append(order, "delivered:"+h); revs = append(revs, rev) },
+ }
+ if err := SyncExecWithCallbacks(ctx, reg, s, pk, dd, "", cb); err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"delivered:inst.example.com", "restart:inst.example.com"}
+ if !slices.Equal(order, want) {
+ t.Fatalf("order = %v, want %v (no report for the sidecar)", order, want)
+ }
+ if !strings.HasPrefix(revs[0], "sha256:") {
+ t.Fatalf("rev = %q, want the rendered identity", revs[0])
+ }
+ order = nil
+ if err := SyncExecWithCallbacks(ctx, reg, s, pk, dd, "", cb); err != nil {
+ t.Fatal(err)
+ }
+ if len(order) != 0 {
+ t.Fatalf("an unchanged file reported %v", order)
+ }
+ if err := SetEnv(ctx, s, "bab", "eleven", map[string]string{"LOG_LEVEL": "info"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := SyncExecWithCallbacks(ctx, reg, s, pk, dd, "", cb); err != nil {
+ t.Fatal(err)
+ }
+ if len(revs) != 2 || revs[1] == revs[0] {
+ t.Fatalf("revs = %v, want a new revision for new content", revs)
+ }
+}
diff --git a/internal/rollout/config.go b/internal/rollout/config.go
new file mode 100644
index 000000000..b978997db
--- /dev/null
+++ b/internal/rollout/config.go
@@ -0,0 +1,76 @@
+package rollout
+
+import (
+ "time"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+)
+
+// configTrack is one instance's two config revisions (spec §8.1). Delivered
+// is what config sync last wrote; active is the revision the current
+// generation started with. They are equal at rest.
+type configTrack struct {
+ delivered string
+ active string
+ // wakeRev is the delivered revision when the wake now in flight was
+ // planned. The wake's spawn reads the file after that, so the revision it
+ // runs is at least this one: activating wakeRev can only under-claim,
+ // which costs one more config attempt, never a missed one.
+ wakeRev string
+ // sleeping: the instance had no process when delivered was written
+ // (spec §8.1). Its next wake starts with that file, so no attempt is
+ // pending and the ack goes out at once, marked Hibernating.
+ sleeping bool
+ fails int
+ lastErr string
+
+ ack *appconfig.ConfigAck
+ ackRev string
+ ackWrite func(appconfig.ConfigAck)
+}
+
+// configBackoff is the wait after the nth consecutive failed config attempt:
+// one minute, doubling, at most thirty (spec §8.1).
+func configBackoff(n int) time.Duration {
+ d := time.Minute
+ for i := 1; i < n && d < 30*time.Minute; i++ {
+ d *= 2
+ }
+ return min(d, 30*time.Minute)
+}
+
+// deferAck holds a config ack until the revision it was written for is
+// active, and returns a func to run now when it already is. A route config
+// sync never reported a delivery for (a shape the scheduler does not
+// restart) acknowledges at once, as before. So does a sleeping instance,
+// with today's hibernating acknowledgement: the config landed and nothing
+// woke to run it (ConfigAck.Hibernating, "delivered on wake").
+func (t *configTrack) deferAck(a appconfig.ConfigAck, write func(appconfig.ConfigAck)) func() {
+ if t.sleeping {
+ a.Hibernating, a.Restarted = true, false
+ return func() { write(a) }
+ }
+ if t.delivered == "" || t.active == t.delivered {
+ return func() { write(a) }
+ }
+ t.ack, t.ackRev, t.ackWrite = &a, t.delivered, write
+ return nil
+}
+
+// activate records rev as active and returns the held ack when it was
+// waiting for exactly rev.
+func (t *configTrack) activate(rev string) (func(), bool) {
+ if rev == "" {
+ return nil, false
+ }
+ t.active = rev
+ if rev == t.delivered {
+ t.fails, t.lastErr = 0, ""
+ }
+ if t.ack == nil || t.ackRev != rev {
+ return nil, false
+ }
+ a, write := *t.ack, t.ackWrite
+ t.ack, t.ackWrite, t.ackRev = nil, nil, ""
+ return func() { write(a) }, true
+}
diff --git a/internal/rollout/config_test.go b/internal/rollout/config_test.go
new file mode 100644
index 000000000..4431920ab
--- /dev/null
+++ b/internal/rollout/config_test.go
@@ -0,0 +1,59 @@
+package rollout
+
+import (
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+)
+
+func TestConfigBackoffRunsFromOneMinuteToThirty(t *testing.T) {
+ want := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute, 16 * time.Minute, 30 * time.Minute, 30 * time.Minute}
+ for i, w := range want {
+ if got := configBackoff(i + 1); got != w {
+ t.Errorf("configBackoff(%d) = %v, want %v", i+1, got, w)
+ }
+ }
+}
+
+func TestAnAckWaitsForItsRevisionToBeActive(t *testing.T) {
+ var wrote []string
+ write := func(a appconfig.ConfigAck) { wrote = append(wrote, a.Host) }
+ tr := &configTrack{delivered: "c2", active: "c1"}
+ if now := tr.deferAck(appconfig.ConfigAck{Host: "a"}, write); now != nil {
+ t.Fatal("acknowledged a delivered revision that is not active")
+ }
+ if run, _ := tr.activate("c1"); run != nil {
+ t.Fatal("an older revision becoming active released the newer ack")
+ }
+ run, ok := tr.activate("c2")
+ if !ok {
+ t.Fatal("the ack for c2 did not come back when c2 became active")
+ }
+ run()
+ if len(wrote) != 1 {
+ t.Fatalf("wrote %v", wrote)
+ }
+ // With nothing delivered through the scheduler (a shape it does not
+ // restart), the ack is written straight away, as today.
+ free := &configTrack{}
+ if now := free.deferAck(appconfig.ConfigAck{Host: "b"}, write); now == nil {
+ t.Fatal("an untracked route's ack must not wait")
+ }
+}
+
+// Spec §8.1: a sleeping instance needs no attempt, because its next wake
+// starts with the delivered file. Its ack goes out at delivery, as today,
+// marked Hibernating like unit config sync's ack for a sleeping route.
+func TestASleepersAckGoesOutAtDeliveryAsHibernating(t *testing.T) {
+ var got []appconfig.ConfigAck
+ tr := &configTrack{delivered: "c2", active: "c2", sleeping: true}
+ run := tr.deferAck(appconfig.ConfigAck{Host: "a", Restarted: true}, func(a appconfig.ConfigAck) { got = append(got, a) })
+ if run == nil {
+ t.Fatal("a sleeper's ack waited for a wake")
+ }
+ run()
+ if len(got) != 1 || !got[0].Hibernating || got[0].Restarted {
+ t.Fatalf("ack = %+v, want Hibernating and not Restarted", got)
+ }
+}