| 1 | // Task 10 of the customer-fleets plan: the two-box fleet acceptance test — |
| 2 | // an edge and TWO fake fleet boxes dialing REAL reverse channels |
| 3 | // ("pi/box-a", "pi/box-b", authenticated out of the registry), driven |
| 4 | // through the whole failure sequence the design spec promises: |
| 5 | // |
| 6 | // (a) a bound route rides its bound box's channel; |
| 7 | // (b) the bound box goes dark -> the survivor serves (rule 2), wakes, |
| 8 | // wins the lapsed lease, publishes a takeover binding, and the |
| 9 | // platform's ReconcileFleetBindings restamps Route.FleetBox to it; |
| 10 | // (c) the dead box comes back, LOSES the reclaim (lease.HeldError) and |
| 11 | // must not steal routing back; |
| 12 | // (d) rotate/detach: rotation refuses NEW dials with the old token |
| 13 | // within one fingerprint poll but does NOT evict the live session; |
| 14 | // detach closes the channel outright; |
| 15 | // (e) both boxes dark -> 503 + Retry-After: 2. |
| 16 | // |
| 17 | // It is `package edge_test`, not `package edge`, and that is forced, not |
| 18 | // stylistic: this test has to drive the REAL binding writer and reconciler |
| 19 | // (internal/host), and internal/host imports internal/edge — the cycle the |
| 20 | // tunnelclient package's own doc comment calls out. An external test package |
| 21 | // may import a package that depends on the package under test, so the file |
| 22 | // stays where the plan puts it (internal/edge/) and pays for it only in a |
| 23 | // few small helpers duplicated from tunnel_e2e_test.go (startFleetBox, |
| 24 | // waitChannel) that cannot be shared across the package boundary. |
| 25 | // |
| 26 | // Hermetic and fast (~10s): loopback httptest edge, real h2 reverse channels |
| 27 | // via the production tunnelclient.Loop, a DirStore for the platform's |
| 28 | // deployment bucket, and an in-test model of the per-host S3 lease. Only the |
| 29 | // LEASE is modeled — see fakeFleetLease — because internal/lease speaks to S3 |
| 30 | // and its short-TTL seam (acquireTTL) is package-private; the live-S3 |
| 31 | // counterpart is TestDarkHostCannotReclaimAfterFailover in |
| 32 | // internal/lease/lease_test.go. Everything else here is the real thing. |
| 33 | package edge_test |
| 34 | |
| 35 | import ( |
| 36 | "context" |
| 37 | "crypto/sha256" |
| 38 | "encoding/hex" |
| 39 | "errors" |
| 40 | "fmt" |
| 41 | "io" |
| 42 | "net/http" |
| 43 | "net/http/httptest" |
| 44 | "net/url" |
| 45 | "path/filepath" |
| 46 | "sync" |
| 47 | "testing" |
| 48 | "time" |
| 49 | |
| 50 | "github.com/carlosframework/platform/internal/edge" |
| 51 | "github.com/carlosframework/platform/internal/host" |
| 52 | "github.com/carlosframework/platform/internal/lease" |
| 53 | "github.com/carlosframework/platform/internal/registry" |
| 54 | "github.com/carlosframework/platform/internal/release" |
| 55 | "github.com/carlosframework/platform/internal/tunnelclient" |
| 56 | ) |
| 57 | |
| 58 | const ( |
| 59 | e2eFleet = "pi" |
| 60 | e2eAppHost = "shop.example" |
| 61 | |
| 62 | // e2eDialBound is how long a fake box gets to complete its first dial (or |
| 63 | // a re-dial after a reap) — the same promise tunnel_e2e_test.go's |
| 64 | // reconnect bound makes, loose enough that a loaded CI box cannot flake |
| 65 | // it. |
| 66 | e2eDialBound = 5 * time.Second |
| 67 | // e2eConvergeBound is the registry->edge convergence promise: the edge |
| 68 | // polls registry.Fingerprint every 2s (Edge.Watch) and reloads routes AND |
| 69 | // fleet channel auth on the same tick, so any registry write — a rotated |
| 70 | // token, a detached box, a restamped FleetBox — is live within one poll. |
| 71 | // 5s is that 2s plus room for a slow reload; a regression to "only on |
| 72 | // restart" (or a fingerprint that misses fleet_boxes.rev) fails here. |
| 73 | e2eConvergeBound = 5 * time.Second |
| 74 | // e2eLeaseTTL models internal/lease's LeaseTTL, compressed from 15 |
| 75 | // minutes to 2 seconds so a whole go-dark -> lapse -> takeover sequence |
| 76 | // fits in a test. The renewer below refreshes at a quarter of it, so a |
| 77 | // box that is ALIVE holds its lease indefinitely and a box that has died |
| 78 | // (its renewer stopped) lapses within one TTL — which is exactly the |
| 79 | // timing property the failover depends on. |
| 80 | e2eLeaseTTL = 2 * time.Second |
| 81 | ) |
| 82 | |
| 83 | // --- the modeled half: the per-host single-writer lease ----------------------- |
| 84 | |
| 85 | // fakeFleetLease models internal/lease's per-host S3 lease: at most one box |
| 86 | // holds a given host at a time, a live hold refuses everyone else with the |
| 87 | // SAME typed *lease.HeldError the real Acquire surfaces (the error the |
| 88 | // activator actually reacts to), and a hold lapses TTL after its last |
| 89 | // renewal. It is a model rather than the real thing because internal/lease |
| 90 | // talks to S3 and its short-TTL seam is package-private; what this test needs |
| 91 | // from it is the ARBITRATION — who may publish a takeover binding — not the |
| 92 | // S3 conditional-write mechanics, which lease_test.go already pins against a |
| 93 | // live bucket. |
| 94 | type fakeFleetLease struct { |
| 95 | ttl time.Duration |
| 96 | mu sync.Mutex |
| 97 | held map[string]fakeLeaseHold |
| 98 | } |
| 99 | |
| 100 | type fakeLeaseHold struct { |
| 101 | box string |
| 102 | expires time.Time |
| 103 | } |
| 104 | |
| 105 | func newFakeFleetLease(ttl time.Duration) *fakeFleetLease { |
| 106 | return &fakeFleetLease{ttl: ttl, held: make(map[string]fakeLeaseHold)} |
| 107 | } |
| 108 | |
| 109 | // acquire claims host for box, or fails with a *lease.HeldError (wrapped, as |
| 110 | // lease.acquireTTL wraps it) if a DIFFERENT box's hold is still live. A box |
| 111 | // re-acquiring its own live hold succeeds, mirroring the real leaser. |
| 112 | func (l *fakeFleetLease) acquire(host, box string) error { |
| 113 | l.mu.Lock() |
| 114 | defer l.mu.Unlock() |
| 115 | if cur, ok := l.held[host]; ok && cur.box != box && time.Now().Before(cur.expires) { |
| 116 | return fmt.Errorf("acquire lease: %w", &lease.HeldError{Owner: cur.box, ExpiresAt: cur.expires}) |
| 117 | } |
| 118 | l.held[host] = fakeLeaseHold{box: box, expires: time.Now().Add(l.ttl)} |
| 119 | return nil |
| 120 | } |
| 121 | |
| 122 | // startRenewer is internal/lease's startRenewer: a live holder keeps pushing |
| 123 | // its expiry out until it stops. Stopping it is how this test says "that box |
| 124 | // just died" — no renewals, so the hold lapses one TTL later and a survivor |
| 125 | // can take over, which is the design spec's failure sequence exactly. |
| 126 | func (l *fakeFleetLease) startRenewer(host, box string) func() { |
| 127 | stop := make(chan struct{}) |
| 128 | done := make(chan struct{}) |
| 129 | go func() { |
| 130 | defer close(done) |
| 131 | t := time.NewTicker(l.ttl / 4) |
| 132 | defer t.Stop() |
| 133 | for { |
| 134 | select { |
| 135 | case <-stop: |
| 136 | return |
| 137 | case <-t.C: |
| 138 | l.mu.Lock() |
| 139 | if cur, ok := l.held[host]; ok && cur.box == box { |
| 140 | l.held[host] = fakeLeaseHold{box: box, expires: time.Now().Add(l.ttl)} |
| 141 | } |
| 142 | l.mu.Unlock() |
| 143 | } |
| 144 | } |
| 145 | }() |
| 146 | var once sync.Once |
| 147 | return func() { |
| 148 | once.Do(func() { close(stop) }) |
| 149 | <-done |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | // --- helpers ------------------------------------------------------------------ |
| 154 | |
| 155 | func e2eHashToken(s string) string { |
| 156 | sum := sha256.Sum256([]byte(s)) |
| 157 | return hex.EncodeToString(sum[:]) |
| 158 | } |
| 159 | |
| 160 | // startFleetBox runs the PRODUCTION host-side dial loop (the same one |
| 161 | // `carlos agent` runs on a Pi) against edgeURL under a fleet channel name, |
| 162 | // and returns a stop func that cancels it and waits — "this box's agent |
| 163 | // died". Mirrors tunnel_e2e_test.go's startFakeRemoteHost, which lives in |
| 164 | // `package edge` and so cannot be reached from here. |
| 165 | func startFleetBox(t *testing.T, edgeURL, name, token string, handler http.Handler) func() { |
| 166 | t.Helper() |
| 167 | ctx, cancel := context.WithCancel(context.Background()) |
| 168 | var wg sync.WaitGroup |
| 169 | wg.Add(1) |
| 170 | go func() { |
| 171 | defer wg.Done() |
| 172 | tunnelclient.Loop(ctx, tunnelclient.Config{ |
| 173 | URL: edgeURL, Name: name, Token: token, Handler: handler, |
| 174 | Logf: func(format string, args ...any) { t.Logf("fake box "+name+": "+format, args...) }, |
| 175 | }) |
| 176 | }() |
| 177 | var once sync.Once |
| 178 | return func() { |
| 179 | once.Do(func() { |
| 180 | cancel() |
| 181 | wg.Wait() // no t.Logf after the test returns |
| 182 | }) |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | // namedFleetHandler answers every request with the box's own label, so a |
| 187 | // response body names the box that actually served it — the only way to tell |
| 188 | // which channel a request rode. |
| 189 | func namedFleetHandler(name string) http.Handler { |
| 190 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 191 | io.WriteString(w, name) |
| 192 | }) |
| 193 | } |
| 194 | |
| 195 | func waitChannel(t *testing.T, hub *edge.TunnelHub, name, want string, within time.Duration) { |
| 196 | t.Helper() |
| 197 | deadline := time.Now().Add(within) |
| 198 | for time.Now().Before(deadline) { |
| 199 | if hub.State(name) == want { |
| 200 | return |
| 201 | } |
| 202 | time.Sleep(20 * time.Millisecond) |
| 203 | } |
| 204 | t.Fatalf("channel %s state = %q, want %q within %s", name, hub.State(name), want, within) |
| 205 | } |
| 206 | |
| 207 | // e2eGet is one client request for the fleet-placed hostname. |
| 208 | func e2eGet(t *testing.T, edgeURL, host string) (int, string) { |
| 209 | t.Helper() |
| 210 | resp, err := e2eDo(edgeURL, host, 10*time.Second) |
| 211 | if err != nil { |
| 212 | t.Fatalf("GET %s (host %s): %v", edgeURL, host, err) |
| 213 | } |
| 214 | defer resp.Body.Close() |
| 215 | body, err := io.ReadAll(resp.Body) |
| 216 | if err != nil { |
| 217 | t.Fatalf("read body: %v", err) |
| 218 | } |
| 219 | return resp.StatusCode, string(body) |
| 220 | } |
| 221 | |
| 222 | func e2eDo(edgeURL, host string, timeout time.Duration) (*http.Response, error) { |
| 223 | req, err := http.NewRequest("GET", edgeURL+"/", nil) |
| 224 | if err != nil { |
| 225 | return nil, err |
| 226 | } |
| 227 | req.Host = host |
| 228 | c := &http.Client{Timeout: timeout} |
| 229 | return c.Do(req) |
| 230 | } |
| 231 | |
| 232 | // waitBody polls until the fleet host answers 200 with want, which is how the |
| 233 | // edge's convergence on a registry change (a restamped FleetBox, applied by |
| 234 | // the 2s Watch poll rebuilding the route's proxy) is observed from outside. |
| 235 | func waitBody(t *testing.T, edgeURL, host, want string, within time.Duration) { |
| 236 | t.Helper() |
| 237 | deadline := time.Now().Add(within) |
| 238 | var lastCode int |
| 239 | var lastBody string |
| 240 | for time.Now().Before(deadline) { |
| 241 | lastCode, lastBody = e2eGet(t, edgeURL, host) |
| 242 | if lastCode == http.StatusOK && lastBody == want { |
| 243 | return |
| 244 | } |
| 245 | time.Sleep(100 * time.Millisecond) |
| 246 | } |
| 247 | t.Fatalf("host %s answered %d %q, want 200 %q within %s", host, lastCode, lastBody, want, within) |
| 248 | } |
| 249 | |
| 250 | // registerProbe asks the hub's /register endpoint the auth question ALONE: it |
| 251 | // presents a channel name and bearer token but deliberately omits the |
| 252 | // Connection/Upgrade headers, so an ACCEPTED credential answers 400 (the |
| 253 | // wire-contract gate, which sits after auth on purpose) and a REJECTED one |
| 254 | // answers 401. That distinction is the whole point — it reads the hub's live |
| 255 | // credential table without hijacking anything, so probing rotation cannot |
| 256 | // disturb the very session whose survival this test also asserts. A probe |
| 257 | // that completed a real handshake would replace the live channel and prove |
| 258 | // the opposite of what it measured. |
| 259 | func registerProbe(t *testing.T, edgeURL, hubHost, name, token string) int { |
| 260 | t.Helper() |
| 261 | req, err := http.NewRequest("GET", edgeURL+"/register", nil) |
| 262 | if err != nil { |
| 263 | t.Fatalf("build probe: %v", err) |
| 264 | } |
| 265 | req.Host = hubHost |
| 266 | req.Header.Set("Carlos-Tunnel-Name", name) |
| 267 | req.Header.Set("Authorization", "Bearer "+token) |
| 268 | resp, err := http.DefaultClient.Do(req) |
| 269 | if err != nil { |
| 270 | t.Fatalf("probe %s: %v", name, err) |
| 271 | } |
| 272 | defer resp.Body.Close() |
| 273 | io.Copy(io.Discard, resp.Body) |
| 274 | return resp.StatusCode |
| 275 | } |
| 276 | |
| 277 | func waitProbe(t *testing.T, edgeURL, hubHost, name, token string, want int, within time.Duration) time.Duration { |
| 278 | t.Helper() |
| 279 | start := time.Now() |
| 280 | deadline := start.Add(within) |
| 281 | var last int |
| 282 | for time.Now().Before(deadline) { |
| 283 | last = registerProbe(t, edgeURL, hubHost, name, token) |
| 284 | if last == want { |
| 285 | return time.Since(start) |
| 286 | } |
| 287 | time.Sleep(100 * time.Millisecond) |
| 288 | } |
| 289 | t.Fatalf("register probe for %s answered %d, want %d within %s", name, last, want, within) |
| 290 | return 0 |
| 291 | } |
| 292 | |
| 293 | // --- the test ----------------------------------------------------------------- |
| 294 | |
| 295 | // TestTwoBoxFleetEndToEnd is the plan's Task 10 acceptance evidence. The legs |
| 296 | // run in sequence against ONE world (each depends on the state the previous |
| 297 | // one left), so it is a single function rather than subtests: a failure means |
| 298 | // the fleet stopped behaving at that step, and nothing after it is meaningful. |
| 299 | // |
| 300 | // Box naming is load-bearing. The fallback rule (rule 2) picks |
| 301 | // ConnectedOfFleet's SORTED FIRST entry, so "box-a" is what a |
| 302 | // fallback-only implementation would always answer: |
| 303 | // |
| 304 | // - Leg (a) binds the route to box-b (sorts LAST) while both boxes are |
| 305 | // connected — only rule 1 can answer "box-b" there. |
| 306 | // - Leg (c) then re-connects box-b while the route is bound to box-a: if |
| 307 | // leg (b)'s restamp had never happened (FleetBox still box-b), or if the |
| 308 | // returning box had stolen the binding back, rule 1 would answer "box-b". |
| 309 | // Insisting on "box-a" with BOTH boxes connected is what makes the |
| 310 | // restamp and the lost reclaim visible rather than assumed. |
| 311 | func TestTwoBoxFleetEndToEnd(t *testing.T) { |
| 312 | ctx := context.Background() |
| 313 | const tokenA, tokenB = "tok-a", "tok-b" |
| 314 | |
| 315 | // --- the platform box: registry, edge, hub, 2s watcher -------------------- |
| 316 | // Deliberately NOT closed at cleanup: Edge.Watch may still be mid-poll when |
| 317 | // the test returns (it exits within one 2s tick), and closing the registry |
| 318 | // under it would race a live reload for no benefit — the temp dir goes away |
| 319 | // either way. |
| 320 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 321 | if err != nil { |
| 322 | t.Fatalf("open registry: %v", err) |
| 323 | } |
| 324 | if err := reg.UpsertFleet(registry.Fleet{Name: e2eFleet, Account: "amadan", Bucket: "pi-data"}); err != nil { |
| 325 | t.Fatalf("UpsertFleet: %v", err) |
| 326 | } |
| 327 | for label, tok := range map[string]string{"box-a": tokenA, "box-b": tokenB} { |
| 328 | if err := reg.AddFleetBox(registry.FleetBox{ |
| 329 | Fleet: e2eFleet, Label: label, TokenHash: e2eHashToken(tok), Status: "active", |
| 330 | }); err != nil { |
| 331 | t.Fatalf("AddFleetBox(%s): %v", label, err) |
| 332 | } |
| 333 | } |
| 334 | // The app: placed on the fleet, bound to box-b (see the naming note above). |
| 335 | if err := reg.Upsert(registry.Route{ |
| 336 | Host: e2eAppHost, Kind: "instance", Status: "ready", |
| 337 | Region: registry.PlacementTarget(e2eFleet), FleetBox: "box-b", |
| 338 | }); err != nil { |
| 339 | t.Fatalf("upsert route: %v", err) |
| 340 | } |
| 341 | |
| 342 | e, err := edge.New(reg, nil) |
| 343 | if err != nil { |
| 344 | t.Fatalf("edge.New: %v", err) |
| 345 | } |
| 346 | srv := httptest.NewServer(e) |
| 347 | t.Cleanup(srv.Close) |
| 348 | u, err := url.Parse(srv.URL) |
| 349 | if err != nil { |
| 350 | t.Fatal(err) |
| 351 | } |
| 352 | // One listener carries both app traffic and channel registration; the hub |
| 353 | // answers on the listener's own hostname (production: CARLOS_TUNNEL_HOST). |
| 354 | hub := edge.NewTunnelHub(u.Hostname(), nil) |
| 355 | e.EnableTunnels(hub) |
| 356 | if err := e.Reload(); err != nil { |
| 357 | t.Fatalf("Reload (wire fleet channel auth): %v", err) |
| 358 | } |
| 359 | // The production watcher, not a hand-driven Reload: every convergence |
| 360 | // bound below is therefore measuring the real 2s fingerprint poll. |
| 361 | stopWatch := make(chan struct{}) |
| 362 | t.Cleanup(func() { close(stopWatch) }) |
| 363 | go e.Watch(stopWatch) |
| 364 | |
| 365 | // --- the fleet: two boxes on real reverse channels ------------------------ |
| 366 | stopA := startFleetBox(t, srv.URL, e2eFleet+"/box-a", tokenA, namedFleetHandler("box-a")) |
| 367 | t.Cleanup(stopA) |
| 368 | stopB := startFleetBox(t, srv.URL, e2eFleet+"/box-b", tokenB, namedFleetHandler("box-b")) |
| 369 | t.Cleanup(stopB) |
| 370 | waitChannel(t, hub, e2eFleet+"/box-a", "connected", e2eDialBound) |
| 371 | waitChannel(t, hub, e2eFleet+"/box-b", "connected", e2eDialBound) |
| 372 | |
| 373 | // The platform's deployment bucket: where fleet boxes publish bindings and |
| 374 | // where the platform box reads them back. One store, both roles — the |
| 375 | // object IS the only channel a box has to say who is serving what. |
| 376 | store := release.NewDirStore(t.TempDir()) |
| 377 | |
| 378 | // box-b is the current holder: it owns the host's lease and has published |
| 379 | // the takeover binding its last wake produced. |
| 380 | leases := newFakeFleetLease(e2eLeaseTTL) |
| 381 | if err := leases.acquire(e2eAppHost, "box-b"); err != nil { |
| 382 | t.Fatalf("box-b initial lease: %v", err) |
| 383 | } |
| 384 | stopRenewB := leases.startRenewer(e2eAppHost, "box-b") |
| 385 | t.Cleanup(stopRenewB) |
| 386 | if err := host.WriteFleetBinding(ctx, store, host.FleetBinding{ |
| 387 | Host: e2eAppHost, Fleet: e2eFleet, Box: "box-b", BoundAt: time.Now().UnixMilli(), |
| 388 | }, true); err != nil { |
| 389 | t.Fatalf("box-b binding: %v", err) |
| 390 | } |
| 391 | |
| 392 | // --- (a) a bound route rides its bound box's channel ---------------------- |
| 393 | if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-b" { |
| 394 | t.Fatalf("(a) got %d %q, want 200 %q — the BOUND box, which sorts last; a fallback-only edge would answer box-a", code, body, "box-b") |
| 395 | } |
| 396 | t.Log("(a) bound route rode the bound box's channel") |
| 397 | |
| 398 | // --- (b) the bound box goes dark ------------------------------------------ |
| 399 | // Its agent stops (power, Wi-Fi, a reboot) — so it renews nothing and |
| 400 | // answers nothing. |
| 401 | stopRenewB() |
| 402 | stopB() |
| 403 | |
| 404 | // The lease is what stops a split brain, and it does not lapse the instant |
| 405 | // the box dies: the survivor's very next wake attempt must LOSE while |
| 406 | // box-b's hold is still live, with the typed error the activator keys off. |
| 407 | err = leases.acquire(e2eAppHost, "box-a") |
| 408 | if err == nil { |
| 409 | t.Fatal("(b) box-a claimed the lease while box-b's hold was still live — that is a split brain, two boxes serving one host") |
| 410 | } |
| 411 | var held *lease.HeldError |
| 412 | if !errors.As(err, &held) { |
| 413 | t.Fatalf("(b) contested acquire returned %T (%v), want a *lease.HeldError", err, err) |
| 414 | } |
| 415 | |
| 416 | // T4 CARRY (observation, not a gate): the edge does not learn a channel |
| 417 | // died from the channel dying. Nothing unregisters a dead conn except |
| 418 | // pingLoop — for a fleet channel that is fleetPingEvery (5s) + |
| 419 | // fleetPingTimeout (5s), so up to ~10s of black hole (spec D7 narrowed it |
| 420 | // from the static channels' ~25s) in which the hub still reports the box |
| 421 | // connected and rule 1 keeps dispatching to a corpse instead of falling |
| 422 | // over to the healthy survivor. That window IS the venue-Wi-Fi failure |
| 423 | // mode: an app stays down for up to ~10s after its box drops off the |
| 424 | // network, even though another box could have served it immediately. |
| 425 | // Observed here without paying the 10s: the assertions below are that the |
| 426 | // channel is STILL registered and the request does NOT reach box-a. |
| 427 | if got := hub.State(e2eFleet + "/box-b"); got != "connected" { |
| 428 | t.Fatalf("(b) hub state for the dead box = %q, want %q — if the hub now reaps eagerly, the black-hole note above is stale and this test should be updated to assert the better behavior", got, "connected") |
| 429 | } |
| 430 | resp, err := e2eDo(srv.URL, e2eAppHost, time.Second) |
| 431 | if err == nil { |
| 432 | defer resp.Body.Close() |
| 433 | body, _ := io.ReadAll(resp.Body) |
| 434 | if resp.StatusCode == http.StatusOK && string(body) == "box-a" { |
| 435 | t.Fatal("(b) the request fell over to box-a before the dead channel was reaped — good news, but it means the ~10s black-hole window this test documents is gone and the comment above needs rewriting") |
| 436 | } |
| 437 | t.Logf("(b) black-hole window: dead-but-registered channel answered %d %q (no fallover)", resp.StatusCode, string(body)) |
| 438 | } else { |
| 439 | t.Logf("(b) black-hole window: dead-but-registered channel hung until the client gave up (%v)", err) |
| 440 | } |
| 441 | |
| 442 | // pingLoop's reap, on demand — CloseChannel is exactly what it calls, so |
| 443 | // the rest of the sequence runs at the speed the fleet would see once the |
| 444 | // ping has fired, not 10s later. |
| 445 | hub.CloseChannel(e2eFleet + "/box-b") |
| 446 | waitChannel(t, hub, e2eFleet+"/box-b", "", time.Second) |
| 447 | |
| 448 | // Rule 2: bound box dark, so any connected box of the fleet serves. The |
| 449 | // app is up again without any registry write at all. |
| 450 | if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-a" { |
| 451 | t.Fatalf("(b) after the bound box went dark: got %d %q, want 200 %q (the fallback rule)", code, body, "box-a") |
| 452 | } |
| 453 | t.Log("(b) fell over to the survivor while still bound to the dead box") |
| 454 | |
| 455 | // The survivor wakes the app locally: it can only claim the lease once |
| 456 | // box-b's hold LAPSES (nothing renews it — the box is dark), and only that |
| 457 | // win authorizes the takeover binding. |
| 458 | acquireDeadline := time.Now().Add(e2eLeaseTTL + 3*time.Second) |
| 459 | var won bool |
| 460 | for time.Now().Before(acquireDeadline) { |
| 461 | if err := leases.acquire(e2eAppHost, "box-a"); err == nil { |
| 462 | won = true |
| 463 | break |
| 464 | } |
| 465 | time.Sleep(50 * time.Millisecond) |
| 466 | } |
| 467 | if !won { |
| 468 | t.Fatalf("(b) box-a never won the lease after box-b's hold should have lapsed (TTL %s)", e2eLeaseTTL) |
| 469 | } |
| 470 | stopRenewA := leases.startRenewer(e2eAppHost, "box-a") |
| 471 | t.Cleanup(stopRenewA) |
| 472 | |
| 473 | // The wake-success publish (fleetBindingHook's shape: takeover=true, |
| 474 | // millisecond BoundAt) followed by the platform's reconcile. In production |
| 475 | // the reconcile rides the agent's instancesLoop; calling it directly is |
| 476 | // what makes this test deterministic instead of loop-timed. |
| 477 | if err := host.WriteFleetBinding(ctx, store, host.FleetBinding{ |
| 478 | Host: e2eAppHost, Fleet: e2eFleet, Box: "box-a", BoundAt: time.Now().UnixMilli(), |
| 479 | }, true); err != nil { |
| 480 | t.Fatalf("(b) box-a takeover binding: %v", err) |
| 481 | } |
| 482 | if err := host.ReconcileFleetBindings(ctx, store, reg); err != nil { |
| 483 | t.Fatalf("(b) ReconcileFleetBindings: %v", err) |
| 484 | } |
| 485 | rt, err := reg.Get(e2eAppHost) |
| 486 | if err != nil { |
| 487 | t.Fatalf("(b) read route: %v", err) |
| 488 | } |
| 489 | if rt.FleetBox != "box-a" { |
| 490 | t.Fatalf("(b) Route.FleetBox = %q, want box-a — the takeover binding never made it back into routing", rt.FleetBox) |
| 491 | } |
| 492 | t.Log("(b) survivor won the lapsed lease, published its takeover binding, and the reconcile restamped FleetBox") |
| 493 | |
| 494 | // --- (c) the dead box returns and loses the reclaim ----------------------- |
| 495 | stopB2 := startFleetBox(t, srv.URL, e2eFleet+"/box-b", tokenB, namedFleetHandler("box-b")) |
| 496 | t.Cleanup(stopB2) |
| 497 | waitChannel(t, hub, e2eFleet+"/box-b", "connected", e2eDialBound) |
| 498 | |
| 499 | // Its activator tries to resume serving the host it used to own. box-a's |
| 500 | // renewer is keeping that lease alive, so the revenant loses — this is the |
| 501 | // HeldError path, the one thing standing between "box-b rebooted" and "two |
| 502 | // boxes writing one SQLite database". |
| 503 | err = leases.acquire(e2eAppHost, "box-b") |
| 504 | if err == nil { |
| 505 | t.Fatal("(c) the returning box reclaimed a lease the survivor is actively renewing — split brain") |
| 506 | } |
| 507 | if !errors.As(err, &held) { |
| 508 | t.Fatalf("(c) contested reclaim returned %T (%v), want a *lease.HeldError", err, err) |
| 509 | } |
| 510 | // No wake success means no takeover write. What box-b's agent DOES run on |
| 511 | // startup is the non-stealing resident pass (publishResidentFleetBindings' |
| 512 | // shape: takeover=false) over the route still sitting in its local |
| 513 | // registry — and that must not claim the binding back. |
| 514 | if err := host.WriteFleetBinding(ctx, store, host.FleetBinding{ |
| 515 | Host: e2eAppHost, Fleet: e2eFleet, Box: "box-b", BoundAt: time.Now().UnixMilli(), |
| 516 | }, false); err != nil { |
| 517 | t.Fatalf("(c) box-b startup pass: %v", err) |
| 518 | } |
| 519 | if err := host.ReconcileFleetBindings(ctx, store, reg); err != nil { |
| 520 | t.Fatalf("(c) ReconcileFleetBindings: %v", err) |
| 521 | } |
| 522 | rt, err = reg.Get(e2eAppHost) |
| 523 | if err != nil { |
| 524 | t.Fatalf("(c) read route: %v", err) |
| 525 | } |
| 526 | if rt.FleetBox != "box-a" { |
| 527 | t.Fatalf("(c) Route.FleetBox = %q, want box-a — the returning box stole routing back without holding the lease", rt.FleetBox) |
| 528 | } |
| 529 | // And end to end, with BOTH boxes connected: the survivor serves. Anything |
| 530 | // that went wrong above — a restamp that never happened, a binding stolen |
| 531 | // back — shows up here as "box-b", because rule 1 would point at box-b in |
| 532 | // either case. Polled, because the restamp only reaches the request path |
| 533 | // through the edge's 2s fingerprint poll. |
| 534 | waitBody(t, srv.URL, e2eAppHost, "box-a", e2eConvergeBound) |
| 535 | t.Log("(c) the returning box lost the reclaim; routing stayed on the survivor with both boxes connected") |
| 536 | |
| 537 | // --- (d) rotation is asymmetric, and converges within one poll ------------ |
| 538 | // Baseline first, so the probe's two answers are known to be |
| 539 | // distinguishable BEFORE the rotation: box-a's current token authorizes |
| 540 | // (400 = authorized, then refused for the missing Upgrade headers). |
| 541 | if code := registerProbe(t, srv.URL, u.Host, e2eFleet+"/box-a", tokenA); code != http.StatusBadRequest { |
| 542 | t.Fatalf("(d) probe baseline = %d, want 400 (credential accepted, wire contract refused)", code) |
| 543 | } |
| 544 | const rotatedA = "tok-a2" |
| 545 | if err := reg.SetFleetBoxToken(e2eFleet, "box-a", e2eHashToken(rotatedA)); err != nil { |
| 546 | t.Fatalf("(d) rotate box-a: %v", err) |
| 547 | } |
| 548 | // Half one: a NEW dial with the OLD token is refused, within one poll. |
| 549 | took := waitProbe(t, srv.URL, u.Host, e2eFleet+"/box-a", tokenA, http.StatusUnauthorized, e2eConvergeBound) |
| 550 | t.Logf("(d) rotation reached the edge's auth table in %s (bound %s)", took, e2eConvergeBound) |
| 551 | // ...and the NEW token is what authorizes now — so the 401 above is a |
| 552 | // rotation, not the channel having vanished from auth altogether (which is |
| 553 | // what a detach does, and which would close the live session below). |
| 554 | if code := registerProbe(t, srv.URL, u.Host, e2eFleet+"/box-a", rotatedA); code != http.StatusBadRequest { |
| 555 | t.Fatalf("(d) probe with the rotated token = %d, want 400 (the new credential must authorize)", code) |
| 556 | } |
| 557 | // Half two, the asymmetry that is the whole design decision: the ALREADY |
| 558 | // LIVE session — still running on the old token, because box-a's agent has |
| 559 | // not been told its credential changed — keeps carrying traffic. If |
| 560 | // rotation closed it, box-a's loop would re-dial with the stale token, |
| 561 | // fail 401 forever, and this request would be answered by box-b instead. |
| 562 | if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-a" { |
| 563 | t.Fatalf("(d) after rotating box-a's token: got %d %q, want 200 %q — rotation must refuse new dials, never evict a live session", code, body, "box-a") |
| 564 | } |
| 565 | t.Log("(d) rotation refused the old token's next dial and left the live session serving") |
| 566 | |
| 567 | // Detach is the OTHER half of that asymmetry: it revokes now. box-b (idle, |
| 568 | // connected, holding nothing) is taken out of the fleet, and its channel |
| 569 | // must be closed by the same Watch->Reload->SetDynamicAuth path. |
| 570 | if err := reg.SetFleetBoxStatus(e2eFleet, "box-b", "detached"); err != nil { |
| 571 | t.Fatalf("(d) detach box-b: %v", err) |
| 572 | } |
| 573 | waitChannel(t, hub, e2eFleet+"/box-b", "", e2eConvergeBound) |
| 574 | // And it STAYS down: its agent re-dials every few hundred ms and is |
| 575 | // refused every time, because a detached box is gone from the auth table |
| 576 | // entirely. A one-shot close that let the box straight back in would pass |
| 577 | // the line above and fail here. 20x100ms (2s, M2 final-review widening |
| 578 | // from 10x100ms/1s) gives a slow CI runner's re-dial loop more than one |
| 579 | // chance to prove it stays refused, not just that it hasn't reconnected |
| 580 | // yet. |
| 581 | for i := 0; i < 20; i++ { |
| 582 | if got := hub.State(e2eFleet + "/box-b"); got != "" { |
| 583 | t.Fatalf("(d) detached box reconnected (state %q) — detach must revoke the credential, not just close the socket", got) |
| 584 | } |
| 585 | time.Sleep(100 * time.Millisecond) |
| 586 | } |
| 587 | // Detaching one box did not disturb the other. |
| 588 | if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-a" { |
| 589 | t.Fatalf("(d) after detaching box-b: got %d %q, want 200 %q", code, body, "box-a") |
| 590 | } |
| 591 | t.Log("(d) detach closed the channel and kept it closed") |
| 592 | |
| 593 | // --- (e) both boxes dark: 503 + Retry-After ------------------------------- |
| 594 | stopA() |
| 595 | hub.CloseChannel(e2eFleet + "/box-a") // pingLoop's reap again |
| 596 | waitChannel(t, hub, e2eFleet+"/box-a", "", time.Second) |
| 597 | if names := hub.ConnectedOfFleet(e2eFleet); len(names) != 0 { |
| 598 | t.Fatalf("(e) fleet still has connected channels %v — the dark case is not set up", names) |
| 599 | } |
| 600 | dark, err := e2eDo(srv.URL, e2eAppHost, 10*time.Second) |
| 601 | if err != nil { |
| 602 | t.Fatalf("(e) GET against a dark fleet: %v", err) |
| 603 | } |
| 604 | defer dark.Body.Close() |
| 605 | if dark.StatusCode != http.StatusServiceUnavailable { |
| 606 | t.Fatalf("(e) dark fleet answered %d, want 503 (not a 502 that reads like a broken backend)", dark.StatusCode) |
| 607 | } |
| 608 | if got := dark.Header.Get("Retry-After"); got != "2" { |
| 609 | t.Fatalf("(e) Retry-After = %q, want %q", got, "2") |
| 610 | } |
| 611 | t.Log("(e) dark fleet answered 503 with Retry-After: 2") |
| 612 | } |
| 613 | |
amadan