carlos / platform Public

Clone
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3// What is testable here without a live S3 is the GATING — who contends, who
4// runs duties, and how the box behaves when it loses or never wins. The lock
5// semantics themselves (a second claimant gets a *lease.HeldError; a dark
6// holder cannot reclaim after failover) belong to internal/lease and are
7// covered there against a real S3-compatible store — internal/lease's own
8// fake (endpoint_test.go's newFakeS3) answers every GET with NoSuchKey, so it
9// proves where the traffic goes and nothing about contention. This file
10// deliberately invents no second fake S3: the election transport is injected
11// as a seam instead.
12
13import (
14 "context"
15 "errors"
16 "fmt"
17 "io"
18 "log"
19 "os"
20 "strings"
21 "sync"
22 "sync/atomic"
23 "testing"
24 "time"
25
26 "github.com/carlosframework/platform/internal/lease"
27)
28
29// ---- helpers ---------------------------------------------------------------
30
31// lockedBuffer is an io.Writer safe for the election goroutine and the test
32// body to touch at once — log output is the assertion in the "stays quiet"
33// tests, so it has to be read while the loop is still running.
34type lockedBuffer struct {
35 mu sync.Mutex
36 b strings.Builder
37}
38
39func (l *lockedBuffer) Write(p []byte) (int, error) {
40 l.mu.Lock()
41 defer l.mu.Unlock()
42 return l.b.Write(p)
43}
44
45func (l *lockedBuffer) String() string {
46 l.mu.Lock()
47 defer l.mu.Unlock()
48 return l.b.String()
49}
50
51// captureRacyLog redirects the standard logger for one test, through a
52// mutexed buffer: unlike this package's other captureLog, the writer here is
53// shared with a live election goroutine.
54func captureRacyLog(t *testing.T) *lockedBuffer {
55 t.Helper()
56 buf := &lockedBuffer{}
57 log.SetOutput(buf)
58 t.Cleanup(func() { log.SetOutput(os.Stderr) })
59 return buf
60}
61
62// quietLog silences the standard logger for a test that doesn't assert on it.
63func quietLog(t *testing.T) {
64 t.Helper()
65 log.SetOutput(io.Discard)
66 t.Cleanup(func() { log.SetOutput(os.Stderr) })
67}
68
69// fakeLease stands in for *lease.Held: renewal health is a func the test
70// flips, and releases are counted.
71type fakeLease struct {
72 mu sync.Mutex
73 err error
74 released int
75}
76
77func (f *fakeLease) fail(err error) {
78 f.mu.Lock()
79 defer f.mu.Unlock()
80 f.err = err
81}
82
83func (f *fakeLease) Err() error {
84 f.mu.Lock()
85 defer f.mu.Unlock()
86 return f.err
87}
88
89func (f *fakeLease) Release(context.Context) error {
90 f.mu.Lock()
91 defer f.mu.Unlock()
92 f.released++
93 return nil
94}
95
96func (f *fakeLease) releases() int {
97 f.mu.Lock()
98 defer f.mu.Unlock()
99 return f.released
100}
101
102type acquireCall struct{ bucket, prefix, region string }
103
104// acquirer records every acquisition attempt and answers from a per-attempt
105// func, so a test can script "contended, contended, won".
106type acquirer struct {
107 mu sync.Mutex
108 calls []acquireCall
109 answer func(n int) (heldLease, error)
110}
111
112func (a *acquirer) acquire(_ context.Context, bucket, prefix, region string) (heldLease, error) {
113 a.mu.Lock()
114 a.calls = append(a.calls, acquireCall{bucket, prefix, region})
115 n := len(a.calls)
116 answer := a.answer
117 a.mu.Unlock()
118 return answer(n)
119}
120
121func (a *acquirer) count() int {
122 a.mu.Lock()
123 defer a.mu.Unlock()
124 return len(a.calls)
125}
126
127func (a *acquirer) first() acquireCall {
128 a.mu.Lock()
129 defer a.mu.Unlock()
130 return a.calls[0]
131}
132
133// contended is what a losing contender sees, wrapped exactly as
134// lease.Acquire wraps it.
135func contended() error {
136 return fmt.Errorf("acquire lease: %w", &lease.HeldError{Owner: "other-box", ExpiresAt: time.Now().Add(time.Minute)})
137}
138
139func waitFor(t *testing.T, what string, cond func() bool) {
140 t.Helper()
141 deadline := time.Now().Add(3 * time.Second)
142 for time.Now().Before(deadline) {
143 if cond() {
144 return
145 }
146 time.Sleep(time.Millisecond)
147 }
148 t.Fatalf("timed out waiting for %s", what)
149}
150
151// runElection arms hk the way RunAgent does and registers its stand-down, so
152// every test exercises the same start/stop pair production uses. The returned
153// func is safe to call twice (the cleanup calls it again).
154func runElection(t *testing.T, hk *housekeeper) func() {
155 t.Helper()
156 stop := hk.start(context.Background())
157 stopped := false
158 once := func() {
159 if stopped {
160 return
161 }
162 stopped = true
163 stop()
164 }
165 t.Cleanup(once)
166 return once
167}
168
169// heldGate is the gate a duty sees on a holder, for the tests that are about
170// duty dispatch rather than the election.
171func heldGate() (context.Context, bool) { return context.Background(), true }
172
173// noGate is the gate every non-holding box sees.
174func noGate() (context.Context, bool) { return nil, false }
175
176func fastHousekeeper(getenv func(string) string, a *acquirer) *housekeeper {
177 hk := newHousekeeper(getenv, a.acquire)
178 hk.retry = time.Millisecond
179 hk.poll = time.Millisecond
180 return hk
181}
182
183// ---- duty gating -----------------------------------------------------------
184
185func TestRunHousekeeperDutiesRunsEachDutyOncePerTickWhenHeld(t *testing.T) {
186 quietLog(t)
187 a, b := 0, 0
188 runHousekeeperDuties(heldGate,
189 housekeeperDuty{"a", func(context.Context) error { a++; return nil }},
190 housekeeperDuty{"b", func(context.Context) error { b++; return nil }},
191 )
192 if a != 1 || b != 1 {
193 t.Fatalf("duty runs = (%d, %d), want (1, 1)", a, b)
194 }
195}
196
197func TestRunHousekeeperDutiesSkipsQuietlyWhenNotHeld(t *testing.T) {
198 buf := captureRacyLog(t)
199 ran := 0
200 runHousekeeperDuties(noGate,
201 housekeeperDuty{"retention pass", func(context.Context) error { ran++; return nil }},
202 )
203 if ran != 0 {
204 t.Fatalf("duty ran %d times on a non-holder, want 0", ran)
205 }
206 // Not holding the lease is the NORMAL state on every box but one, every
207 // tick, forever: a log line here would bury the log.
208 if out := buf.String(); out != "" {
209 t.Fatalf("non-holder logged %q, want silence", out)
210 }
211}
212
213func TestRunHousekeeperDutiesLogsOneDutysErrorAndRunsTheRest(t *testing.T) {
214 buf := captureRacyLog(t)
215 later := 0
216 runHousekeeperDuties(heldGate,
217 housekeeperDuty{"retention pass", func(context.Context) error { return errors.New("boom") }},
218 housekeeperDuty{"later duty", func(context.Context) error { later++; return nil }},
219 )
220 if later != 1 {
221 t.Fatalf("later duty ran %d times, want 1 — one duty's error must not stop the next", later)
222 }
223 out := buf.String()
224 if !strings.Contains(out, "retention pass") || !strings.Contains(out, "boom") {
225 t.Fatalf("log = %q, want the failing duty's name and error", out)
226 }
227}
228
229// ---- who contends ----------------------------------------------------------
230
231func TestHousekeeperFleetBoxNeverAttemptsAcquisition(t *testing.T) {
232 quietLog(t)
233 a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
234 hk := fastHousekeeper(envFunc(map[string]string{
235 "CARLOS_FLEET": "acme",
236 "CARLOS_DEPLOYMENT_BUCKET": "dep-bucket",
237 }), a)
238
239 // run returns immediately on a fleet box — nothing to loop over.
240 hk.run(context.Background())
241
242 if a.count() != 0 {
243 t.Fatalf("fleet box attempted acquisition %d times, want 0", a.count())
244 }
245 if hk.held() {
246 t.Fatal("fleet box reports holding the housekeeper lease")
247 }
248}
249
250func TestHousekeeperWithNoDeploymentStoreNeverHolds(t *testing.T) {
251 quietLog(t)
252 a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
253 hk := fastHousekeeper(envFunc(map[string]string{}), a)
254
255 hk.run(context.Background())
256
257 if a.count() != 0 {
258 t.Fatalf("storeless box attempted acquisition %d times, want 0", a.count())
259 }
260 if hk.held() {
261 t.Fatal("storeless box reports holding the housekeeper lease")
262 }
263}
264
265func TestHousekeeperDirDeploymentIsItsOwnHousekeeper(t *testing.T) {
266 quietLog(t)
267 a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
268 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": t.TempDir()}), a)
269 stop := runElection(t, hk)
270
271 // No goroutine to wait on: a lone box holds from the instant it is armed,
272 // because there is nothing to acquire.
273 if !hk.held() {
274 t.Fatal("dir deployment does not hold: a directory-backed deployment is one box by construction and must keep pruning")
275 }
276 if a.count() != 0 {
277 t.Fatalf("dir deployment attempted acquisition %d times, want 0 — there is no bucket to elect in", a.count())
278 }
279
280 // And it stands down with the process, so nothing believes it is the
281 // housekeeper after the agent has gone.
282 stop()
283 if hk.held() {
284 t.Fatal("dir deployment still reports holding after stand-down")
285 }
286}
287
288// ---- the election ----------------------------------------------------------
289
290func TestHousekeeperAcquiresTheDeploymentLeaseAndHolds(t *testing.T) {
291 quietLog(t)
292 a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
293 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
294 runElection(t, hk)
295
296 waitFor(t, "the lease to be held", hk.held)
297
298 got := a.first()
299 want := acquireCall{bucket: "dep-bucket", prefix: housekeeperLeasePrefix, region: ""}
300 if got != want {
301 t.Fatalf("acquire call = %+v, want %+v", got, want)
302 }
303 if housekeeperLeasePrefix != "control/housekeeper-lease" {
304 t.Fatalf("lease prefix = %q, want control/housekeeper-lease (lock.json lands under it)", housekeeperLeasePrefix)
305 }
306}
307
308func TestHousekeeperLosingTheElectionIsQuietAndKeepsTrying(t *testing.T) {
309 buf := captureRacyLog(t)
310 a := &acquirer{answer: func(int) (heldLease, error) { return nil, contended() }}
311 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
312 // The arming line is not what this test is about; drop everything logged
313 // before the loop starts contending.
314 runElection(t, hk)
315
316 waitFor(t, "several acquisition attempts", func() bool { return a.count() >= 3 })
317
318 if hk.held() {
319 t.Fatal("a losing contender reports holding the lease")
320 }
321 // Contention is normal on every tick a box doesn't win: an error loop
322 // here would be a line every retry, forever, on N-1 boxes.
323 for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {
324 if strings.Contains(line, "lease acquire") {
325 t.Fatalf("contention logged as a failure: %q", line)
326 }
327 }
328}
329
330func TestHousekeeperLogsARealAcquireFailure(t *testing.T) {
331 buf := captureRacyLog(t)
332 a := &acquirer{answer: func(int) (heldLease, error) { return nil, errors.New("no credentials") }}
333 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
334 runElection(t, hk)
335
336 waitFor(t, "the failure to be logged", func() bool { return strings.Contains(buf.String(), "no credentials") })
337
338 if hk.held() {
339 t.Fatal("a box that could not acquire reports holding the lease")
340 }
341}
342
343func TestHousekeeperStandsDownWhenRenewalFails(t *testing.T) {
344 quietLog(t)
345 first := &fakeLease{}
346 third := &fakeLease{}
347 // The takeover, scripted: this box holds, loses renewal, and finds the
348 // lock already taken by whoever won it after expiry — for as long as the
349 // test says so. That middle state is what makes "stood down" observable
350 // at all: if a lease were waiting on the very next attempt, the box would
351 // be holding again within a retry and a housekeeper that never cleared
352 // its flag would look identical to one that did.
353 var freed atomic.Bool
354 a := &acquirer{answer: func(n int) (heldLease, error) {
355 switch {
356 case n == 1:
357 return first, nil
358 case !freed.Load():
359 return nil, contended()
360 default:
361 return third, nil
362 }
363 }}
364 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
365 runElection(t, hk)
366
367 waitFor(t, "the first lease to be held", hk.held)
368
369 // Renewal fails: leadership is lost, so this box must stop reporting
370 // itself as the housekeeper — a stale holder still running duties (or,
371 // from Part 2, still issuing certs) against a lease another box now owns
372 // is the whole reason this check exists.
373 first.fail(errors.New("renew: precondition failed"))
374 waitFor(t, "the stand-down", func() bool { return !hk.held() })
375 waitFor(t, "a fresh acquisition attempt", func() bool { return a.count() > 1 })
376
377 // A lost lease is NOT released: our etag is stale, and the TTL is what
378 // hands it over.
379 if first.releases() != 0 {
380 t.Fatalf("released a lost lease %d times, want 0", first.releases())
381 }
382
383 // And it recovers on its own once the lock frees up again.
384 freed.Store(true)
385 waitFor(t, "the lease to be won back", hk.held)
386 if third.releases() != 0 {
387 t.Fatalf("released the recovered lease %d times, want 0 while still holding it", third.releases())
388 }
389}
390
391// The stand-down is the path an ordinary `systemctl restart` takes, so it has
392// to release the lock AND come back — a stop that hung would wedge the
393// restart, and one that returned before the loop had released would strand
394// control/housekeeper-lease/lock.json for its 15-minute TTL.
395func TestHousekeeperStopReleasesTheLeaseAndReturns(t *testing.T) {
396 quietLog(t)
397 held := &fakeLease{}
398 a := &acquirer{answer: func(int) (heldLease, error) { return held, nil }}
399 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
400 stop := runElection(t, hk)
401
402 waitFor(t, "the lease to be held", hk.held)
403
404 returned := make(chan struct{})
405 go func() {
406 stop()
407 close(returned)
408 }()
409 select {
410 case <-returned:
411 case <-time.After(3 * time.Second):
412 t.Fatal("stop() did not return — a shutdown would hang here")
413 }
414
415 if hk.held() {
416 t.Fatal("still reports holding the lease after shutdown")
417 }
418 if held.releases() != 1 {
419 t.Fatalf("releases = %d, want 1 — a graceful stand-down hands the lease over immediately", held.releases())
420 }
421 // The election goroutine is gone, not merely told to go: stop() waits on
422 // it, so a leaked loop would still be acquiring after this point.
423 before := a.count()
424 time.Sleep(20 * time.Millisecond) // many retries at the 1ms test cadence
425 if after := a.count(); after != before {
426 t.Fatalf("acquisitions after stop: %d → %d, want no further attempts (the loop leaked)", before, after)
427 }
428}
429
430// ---- duties are bound to the holding ---------------------------------------
431
432// A duty must not outlive the authority it started under: when renewal fails
433// mid-prune, the sweep's context is cancelled rather than left to finish its
434// deletes against a lock another box now owns.
435func TestHousekeeperCancelsDutiesWhenTheHoldEnds(t *testing.T) {
436 quietLog(t)
437 held := &fakeLease{}
438 a := &acquirer{answer: func(n int) (heldLease, error) {
439 if n == 1 {
440 return held, nil
441 }
442 return nil, contended()
443 }}
444 hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
445 runElection(t, hk)
446 waitFor(t, "the lease to be held", hk.held)
447
448 started := make(chan struct{})
449 observed := make(chan error, 1)
450 go func() {
451 runHousekeeperDuties(hk.gate, housekeeperDuty{"slow duty", func(ctx context.Context) error {
452 close(started)
453 select {
454 case <-ctx.Done():
455 observed <- ctx.Err()
456 case <-time.After(3 * time.Second):
457 observed <- nil
458 }
459 return ctx.Err()
460 }})
461 }()
462
463 <-started
464 held.fail(errors.New("renew: precondition failed"))
465
466 select {
467 case err := <-observed:
468 if err == nil {
469 t.Fatal("the duty ran to completion after the hold ended — it must be cancelled")
470 }
471 case <-time.After(3 * time.Second):
472 t.Fatal("timed out waiting for the duty to observe the stand-down")
473 }
474
475 // And the gate is shut for the next tick, so nothing new starts either.
476 if _, ok := hk.gate(); ok {
477 t.Fatal("the gate is still open after the hold ended")
478 }
479}
480
481// A duty that starts on a holder and is cancelled mid-flight stood down; it
482// did not fail, so it must not be logged as a failure — and the duties behind
483// it in the same tick are no longer this box's to run.
484func TestRunHousekeeperDutiesStopQuietlyWhenTheHoldEndsMidTick(t *testing.T) {
485 buf := captureRacyLog(t)
486 ctx, cancel := context.WithCancel(context.Background())
487 later := 0
488 runHousekeeperDuties(
489 func() (context.Context, bool) { return ctx, true },
490 housekeeperDuty{"retention pass", func(ctx context.Context) error {
491 cancel() // the hold ends while this duty is running
492 return ctx.Err()
493 }},
494 housekeeperDuty{"later duty", func(context.Context) error { later++; return nil }},
495 )
496 if later != 0 {
497 t.Fatalf("later duty ran %d times after the hold ended, want 0", later)
498 }
499 if out := buf.String(); out != "" {
500 t.Fatalf("stand-down logged as a failure: %q", out)
501 }
502}
503
504// ---- the process-wide gate -------------------------------------------------
505
506func TestHousekeeperHeldIsFalseBeforeAnyElectionIsArmed(t *testing.T) {
507 if housekeeperHeld() {
508 t.Fatal("housekeeperHeld() is true with no housekeeper armed")
509 }
510 if _, ok := housekeeperGate(); ok {
511 t.Fatal("housekeeperGate() reports held with no housekeeper armed")
512 }
513}
514
515// ---- mintMeshKeyLoop (multi-region backhaul spec, Part 3 Task 2's review
516// fix): the mesh key's own arming loop, independent of retentionLoop -------
517
518// Mirrors TestRetentionLoopIsAPlatformBoxJob's shape (agent_test.go): a
519// fleet box must skip before ever opening a store.
520func TestMintMeshKeyLoopIsAPlatformBoxJob(t *testing.T) {
521 const skipLine = "agent: mesh key mint loop is a platform-box job — skipped on a fleet box"
522
523 logs := captureLog(t)
524 door := newDoorFake(t)
525 for k, v := range fleetDoorEnv(door.url) {
526 t.Setenv(k, v)
527 }
528 t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
529 t.Setenv("CARLOS_DEPLOYMENT_DIR", "")
530
531 mintMeshKeyLoop() // must return; a fleet box never reaches the loop
532
533 if got := logs(); !strings.Contains(got, skipLine) {
534 t.Fatalf("log = %q, want the skip line verbatim", got)
535 }
536 if hits := door.seen(); len(hits) != 0 {
537 t.Fatalf("door saw %+v, want zero — the skip must land BEFORE the store opens", hits)
538 }
539}
540
541// The regression guard for the Critical review finding: the mint must NOT be
542// gated on CARLOS_RELEASE_PUBKEY (retentionLoop's own gate, unrelated to the
543// mesh key) — a box missing that var still wins the housekeeper election, so
544// a mint duty tied to retentionLoop's arming would silently never run
545// fleet-wide. Proven by explicitly leaving CARLOS_RELEASE_PUBKEY unset and
546// confirming the loop still arms.
547func TestMintMeshKeyLoopArmsWithoutReleasePubkey(t *testing.T) {
548 t.Setenv("CARLOS_FLEET", "")
549 t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
550 t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())
551 t.Setenv("CARLOS_RELEASE_PUBKEY", "") // deliberately unset — the whole point
552
553 buf := captureRacyLog(t)
554 // mintMeshKeyLoop never returns once armed: run it in its own goroutine
555 // (it parks in a time.Sleep for a minute, long past this test's end) and
556 // wait for the arming log line against a synchronized sink.
557 go mintMeshKeyLoop()
558
559 waitFor(t, "the mint loop to arm despite no CARLOS_RELEASE_PUBKEY", func() bool {
560 return strings.Contains(buf.String(), "agent: mesh key mint loop armed")
561 })
562}
563