| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "io" |
| 6 | "net/http" |
| 7 | "os" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "sync/atomic" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/instancectl" |
| 15 | "github.com/carlosframework/platform/internal/schedule" |
| 16 | ) |
| 17 | |
| 18 | // The switch retires exactly the generation it switched away from (spec |
| 19 | // §7.10), and the retirement is latched (final review F8): the retiring |
| 20 | // generation's writes stay refused after it has drained and left the |
| 21 | // operation set, since a handler that began before the switch may still be |
| 22 | // running. The successor's writes apply throughout, and a hibernating stop |
| 23 | // after the swap retires nothing. |
| 24 | func TestRetiredGenNamesOnlyTheRetiringGeneration(t *testing.T) { |
| 25 | t.Parallel() |
| 26 | f := newSwapFixture(t) |
| 27 | duringHealth := map[string]bool{} |
| 28 | f.a.genHealth = func(_ context.Context, g Gen) error { |
| 29 | duringHealth["successor"] = f.a.RetiredGen(f.host, g.N) |
| 30 | duringHealth["incumbent"] = f.a.RetiredGen(f.host, f.incumbent.N) |
| 31 | return nil |
| 32 | } |
| 33 | gate := make(chan struct{}) |
| 34 | f.b.lastProc().setStopGate(gate) // hold the retiring generation's drain open |
| 35 | results := make(chan SwapResult, 1) |
| 36 | go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| 37 | waitFor(t, "the switch", func() bool { return indexPrefix(f.b.events(), "switch:") >= 0 }) |
| 38 | if duringHealth["successor"] || duringHealth["incumbent"] { |
| 39 | t.Fatalf("before the switch: %v, want both open", duringHealth) |
| 40 | } |
| 41 | succ, _ := f.b.lastGen() |
| 42 | if !f.a.RetiredGen(f.host, f.incumbent.N) { |
| 43 | t.Fatal("the retiring generation's schedule writes are still open after the switch") |
| 44 | } |
| 45 | if f.a.RetiredGen(f.host, succ.N) { |
| 46 | t.Fatal("the current generation's writes are refused") |
| 47 | } |
| 48 | if f.a.ScheduleWrite(f.host, f.incumbent.N, func() { t.Fatal("a retired generation's write ran") }) { |
| 49 | t.Fatal("ScheduleWrite ran the retired generation's write") |
| 50 | } |
| 51 | close(gate) |
| 52 | if res := <-results; res.Outcome != SwapDone { |
| 53 | t.Fatalf("Swap = %+v", res) |
| 54 | } |
| 55 | if _, members := f.a.genStateForTest(f.host); len(members) != 0 { |
| 56 | t.Fatalf("members %v: the retiring generation should have left the set", members) |
| 57 | } |
| 58 | if !f.a.RetiredGen(f.host, f.incumbent.N) { |
| 59 | t.Fatal("the retirement was not latched: the drained generation's writes reopened") |
| 60 | } |
| 61 | ran := false |
| 62 | if !f.a.ScheduleWrite(f.host, succ.N, func() { ran = true }) || !ran { |
| 63 | t.Fatal("the successor's write was refused") |
| 64 | } |
| 65 | // A hibernating stop is unchanged: the stopping generation is current. |
| 66 | stopGate := make(chan struct{}) |
| 67 | _, cur := f.b.lastGen() |
| 68 | cur.setStopGate(stopGate) |
| 69 | f.clk.advance(2 * time.Hour) |
| 70 | f.a.sweepOnce() |
| 71 | waitFor(t, "the hibernating stop", func() bool { |
| 72 | f.a.mu.Lock() |
| 73 | defer f.a.mu.Unlock() |
| 74 | e := f.a.entries[f.host] |
| 75 | return e != nil && e.state == actStopping |
| 76 | }) |
| 77 | if f.a.RetiredGen(f.host, succ.N) { |
| 78 | t.Fatal("a hibernating generation's writes are refused") |
| 79 | } |
| 80 | close(stopGate) |
| 81 | } |
| 82 | |
| 83 | // schedSwitchWaiting reports whether a switch holds host's write fence and |
| 84 | // is waiting for a write in flight. |
| 85 | func (a *Service) schedSwitchWaiting(host string) bool { |
| 86 | f := a.schedFenceFor(host) |
| 87 | f.mu.Lock() |
| 88 | defer f.mu.Unlock() |
| 89 | return f.switching && f.writes > 0 |
| 90 | } |
| 91 | |
| 92 | // Final review F8: the switch is serialized against the schedule writes. |
| 93 | // A write of the incumbent that is under way when the switch comes holds |
| 94 | // the switch until it finishes (so it lands before the switch, while its |
| 95 | // generation is current), and once the switch has happened every write of |
| 96 | // the retired generation is refused. |
| 97 | func TestScheduleWriteIsSerializedWithTheSwitch(t *testing.T) { |
| 98 | t.Parallel() |
| 99 | f := newSwapFixture(t) |
| 100 | entered, release := make(chan struct{}), make(chan struct{}) |
| 101 | wrote := make(chan bool, 1) |
| 102 | go func() { |
| 103 | wrote <- f.a.ScheduleWrite(f.host, f.incumbent.N, func() { |
| 104 | close(entered) |
| 105 | <-release |
| 106 | f.b.record("schedule-write") |
| 107 | }) |
| 108 | }() |
| 109 | <-entered |
| 110 | results := make(chan SwapResult, 1) |
| 111 | go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| 112 | waitFor(t, "the switch waiting on the write", func() bool { return f.a.schedSwitchWaiting(f.host) }) |
| 113 | if ev := f.b.events(); indexPrefix(ev, "switch:") >= 0 { |
| 114 | t.Fatalf("events %v: the switch ran while a schedule write was under way", ev) |
| 115 | } |
| 116 | close(release) |
| 117 | if !<-wrote { |
| 118 | t.Fatal("the incumbent's write, begun before the switch, was refused") |
| 119 | } |
| 120 | if res := <-results; res.Outcome != SwapDone { |
| 121 | t.Fatalf("Swap = %+v", res) |
| 122 | } |
| 123 | ev := f.b.events() |
| 124 | if w, sw := indexPrefix(ev, "schedule-write"), indexPrefix(ev, "switch:"); w < 0 || sw < 0 || w > sw { |
| 125 | t.Fatalf("events %v: want the write before the switch", ev) |
| 126 | } |
| 127 | if f.a.ScheduleWrite(f.host, f.incumbent.N, func() { t.Fatal("a retired generation's write ran") }) { |
| 128 | t.Fatal("the retired generation's write after the switch was not refused") |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | // Flag off: nothing is retired and nothing is serialized. A generation the |
| 133 | // fence has latched (here by hand, as a switch would with swaps on) still |
| 134 | // writes once the switch is off. |
| 135 | func TestScheduleWriteIsUnfencedWithSwapOff(t *testing.T) { |
| 136 | f := newSwapFixture(t) |
| 137 | f.a.schedFenceFor(f.host).retire(77) |
| 138 | if f.a.ScheduleWrite(f.host, 77, func() {}) || !f.a.RetiredGen(f.host, 77) { |
| 139 | t.Fatal("with swaps on, a latched generation's write was not refused") |
| 140 | } |
| 141 | f.a.gens = false |
| 142 | ran := false |
| 143 | if !f.a.ScheduleWrite(f.host, 77, func() { ran = true }) || !ran || f.a.RetiredGen(f.host, 77) { |
| 144 | t.Fatal("with swaps off, a write was refused or a generation reported retired") |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | func TestExecBackendWriteFence(t *testing.T) { |
| 149 | b := testExecBackend(t) |
| 150 | if b.writeFence("h.example.com", 7) != nil { |
| 151 | t.Fatal("a backend with no ScheduleFence seam fenced the schedule verbs") |
| 152 | } |
| 153 | var gotHost string |
| 154 | var gotGen uint64 |
| 155 | b.ScheduleFence = func(h string, g uint64, write func()) bool { gotHost, gotGen = h, g; write(); return true } |
| 156 | ran := false |
| 157 | if !b.writeFence("h.example.com", 7)(func() { ran = true }) || !ran || gotHost != "h.example.com" || gotGen != 7 { |
| 158 | t.Fatalf("fence asked %q %d, ran %v", gotHost, gotGen, ran) |
| 159 | } |
| 160 | } |
| 161 | |
| 162 | // The seam reaches the real control socket of a spawned child (the flat |
| 163 | // listener): the verbs are mounted through the fence, the fence is asked |
| 164 | // about this host and the generation the listener was bound for (0 for a |
| 165 | // plain Start), and once it refuses a write is a 409 while a read still |
| 166 | // answers. |
| 167 | func TestExecBackendStartGatesTheControlSocketWrites(t *testing.T) { |
| 168 | b := testExecBackend(t) |
| 169 | b.Timers = &fakeTimerStore{} |
| 170 | var retired atomic.Bool |
| 171 | var askedHost atomic.Value |
| 172 | askedGen := uint64(99) |
| 173 | var mu sync.Mutex |
| 174 | b.ScheduleFence = func(host string, gen uint64, write func()) bool { |
| 175 | askedHost.Store(host) |
| 176 | mu.Lock() |
| 177 | askedGen = gen |
| 178 | mu.Unlock() |
| 179 | if retired.Load() { |
| 180 | return false |
| 181 | } |
| 182 | write() |
| 183 | return true |
| 184 | } |
| 185 | ctl, token, stop := startAndReachControlSocket(t, b, "gated.example.com") |
| 186 | defer stop() |
| 187 | |
| 188 | send := func(method, path, body string) (int, string) { |
| 189 | req, err := http.NewRequest(method, "http://carlos"+path, strings.NewReader(body)) |
| 190 | if err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | req.Header.Set("Authorization", "Bearer "+token) |
| 194 | resp, err := socketClient(ctl).Do(req) |
| 195 | if err != nil { |
| 196 | t.Fatalf("dial control socket %s: %v", ctl, err) |
| 197 | } |
| 198 | defer resp.Body.Close() |
| 199 | var sb strings.Builder |
| 200 | buf := make([]byte, 512) |
| 201 | for { |
| 202 | n, err := resp.Body.Read(buf) |
| 203 | sb.Write(buf[:n]) |
| 204 | if err != nil { |
| 205 | break |
| 206 | } |
| 207 | } |
| 208 | return resp.StatusCode, sb.String() |
| 209 | } |
| 210 | if code, _ := send("DELETE", "/schedule/remind", ""); code != http.StatusNoContent { |
| 211 | t.Fatalf("open DELETE = %d, want 204", code) |
| 212 | } |
| 213 | retired.Store(true) |
| 214 | if code, msg := send("DELETE", "/schedule/remind", ""); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) { |
| 215 | t.Fatalf("retired DELETE = %d %q, want 409", code, msg) |
| 216 | } |
| 217 | if code, msg := send("PUT", "/schedule/remind", `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) { |
| 218 | t.Fatalf("retired PUT = %d %q, want 409", code, msg) |
| 219 | } |
| 220 | if code, _ := send("GET", "/schedule", ""); code != http.StatusOK { |
| 221 | t.Fatalf("retired GET = %d, want 200", code) |
| 222 | } |
| 223 | mu.Lock() |
| 224 | gen := askedGen |
| 225 | mu.Unlock() |
| 226 | if askedHost.Load() != "gated.example.com" || gen != 0 { |
| 227 | t.Fatalf("fence asked %v generation %d, want gated.example.com generation 0", askedHost.Load(), gen) |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | // Final review F8, T8: the nested layout's control listener (prepareAt, |
| 232 | // which the nested generation start uses) is mounted through the same |
| 233 | // fence, asked for the generation it was prepared for. |
| 234 | func TestPreparedControlListenerIsFenced(t *testing.T) { |
| 235 | b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{}) |
| 236 | b.GenContainment = true |
| 237 | b.Timers = &fakeTimerStore{} |
| 238 | var retired atomic.Bool |
| 239 | var mu sync.Mutex |
| 240 | asked := map[uint64]int{} |
| 241 | b.ScheduleFence = func(host string, gen uint64, write func()) bool { |
| 242 | mu.Lock() |
| 243 | if host == rt.Host { |
| 244 | asked[gen]++ |
| 245 | } |
| 246 | mu.Unlock() |
| 247 | if retired.Load() { |
| 248 | return false |
| 249 | } |
| 250 | write() |
| 251 | return true |
| 252 | } |
| 253 | if l, _ := b.Layout(rt); l != LayoutNested { |
| 254 | t.Fatalf("layout %v, want nested", l) |
| 255 | } |
| 256 | p0 := b.pathsFor(rt) |
| 257 | if err := os.MkdirAll(p0.runDir, 0o750); err != nil { |
| 258 | t.Fatal(err) |
| 259 | } |
| 260 | sock := nestedGenSocketIn(p0.runDir, 7) |
| 261 | p, ctl, logs, _, err := b.prepareAt(rt, instancectl.RoleRun, sock, 7) |
| 262 | if err != nil { |
| 263 | t.Fatalf("prepareAt: %v", err) |
| 264 | } |
| 265 | if logs != nil { |
| 266 | defer logs.Close() |
| 267 | } |
| 268 | if ctl == nil { |
| 269 | t.Fatal("prepareAt bound no control listener") |
| 270 | } |
| 271 | defer ctl.Close() |
| 272 | env, err := readEnvFileBeneath(p.dataDir, rt.Host+".env") |
| 273 | if err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | token := env["CARLOS_ADMIN_TOKEN"] |
| 277 | send := func(method, path, body string) (int, string) { |
| 278 | req, err := http.NewRequest(method, "http://carlos"+path, strings.NewReader(body)) |
| 279 | if err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | req.Header.Set("Authorization", "Bearer "+token) |
| 283 | resp, err := socketClient(sock + ".control").Do(req) |
| 284 | if err != nil { |
| 285 | t.Fatalf("dial control socket: %v", err) |
| 286 | } |
| 287 | defer resp.Body.Close() |
| 288 | buf, _ := io.ReadAll(resp.Body) |
| 289 | return resp.StatusCode, string(buf) |
| 290 | } |
| 291 | if code, msg := send("DELETE", "/schedule/remind", ""); code != http.StatusNoContent { |
| 292 | t.Fatalf("open DELETE = %d %q, want 204", code, msg) |
| 293 | } |
| 294 | retired.Store(true) |
| 295 | if code, msg := send("PUT", "/schedule/remind", `{"at":"2026-09-02T08:00:00Z","path":"/r2"}`); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) { |
| 296 | t.Fatalf("retired PUT = %d %q, want 409", code, msg) |
| 297 | } |
| 298 | if code, _ := send("DELETE", "/schedule/remind", ""); code != http.StatusConflict { |
| 299 | t.Fatalf("retired DELETE = %d, want 409", code) |
| 300 | } |
| 301 | if code, _ := send("GET", "/schedule", ""); code != http.StatusOK { |
| 302 | t.Fatalf("retired GET = %d, want 200", code) |
| 303 | } |
| 304 | mu.Lock() |
| 305 | defer mu.Unlock() |
| 306 | if asked[7] != 3 || len(asked) != 1 { |
| 307 | t.Fatalf("fence asked %v, want generation 7 for each of the three writes", asked) |
| 308 | } |
| 309 | } |
| 310 | |
amadan