| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "net/http" |
| 7 | "net/http/httptest" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/registry" |
| 12 | ) |
| 13 | |
| 14 | // Issue #213: a box that cannot provide an instance's config must not take |
| 15 | // its lease. The gate runs inside ExecBackend.Acquire, before lease.Acquire, |
| 16 | // so every wake source (request, tick, sweeper, worker boot) is covered by |
| 17 | // the one place a lease is taken. |
| 18 | |
| 19 | // The gate is asked about the route being woken, and a refusal means the |
| 20 | // lease is never attempted. Cfg.Bucket is empty here, so a call that fell |
| 21 | // through to lease.Acquire would fail with a different error. |
| 22 | func TestExecBackendAcquireRefusesBeforeTheLease(t *testing.T) { |
| 23 | var asked []registry.Route |
| 24 | boom := errors.New("cannot open sealed config") |
| 25 | b := &ExecBackend{ConfigGate: func(_ context.Context, rt registry.Route) error { |
| 26 | asked = append(asked, rt) |
| 27 | return boom |
| 28 | }} |
| 29 | rt := registry.Route{Host: "kpmg.example.com", Kind: "instance", App: "shop", Hibernate: true} |
| 30 | |
| 31 | lease, err := b.Acquire(context.Background(), rt) |
| 32 | if lease != nil { |
| 33 | t.Fatalf("lease = %v, want nil", lease) |
| 34 | } |
| 35 | var refused *errConfigRefused |
| 36 | if !errors.As(err, &refused) { |
| 37 | t.Fatalf("err = %v, want *errConfigRefused", err) |
| 38 | } |
| 39 | if !errors.Is(err, boom) { |
| 40 | t.Errorf("err %v does not wrap the gate's reason", err) |
| 41 | } |
| 42 | if len(asked) != 1 || asked[0].Host != rt.Host { |
| 43 | t.Errorf("gate asked about %v, want exactly %s", asked, rt.Host) |
| 44 | } |
| 45 | } |
| 46 | |
| 47 | // The refusal is the member's 503 with a Retry-After, answered quietly (the |
| 48 | // guard logs once per host; a log line per request would drown it), and the |
| 49 | // failed wake leaves no entry behind so the next request asks again. |
| 50 | func TestActivatorConfigRefusal503(t *testing.T) { |
| 51 | a, b, _, reg := testActivator(t, time.Minute) |
| 52 | b.acquireErr = &errConfigRefused{err: errors.New("no key")} |
| 53 | |
| 54 | if err := reg.Upsert(registry.Route{Host: "nokey.example.com", Kind: "instance", Socket: "/tmp/nokey.sock", Hibernate: true}); err != nil { |
| 55 | t.Fatalf("upsert: %v", err) |
| 56 | } |
| 57 | rt, err := newTestRouter(reg, a) |
| 58 | if err != nil { |
| 59 | t.Fatalf("new router: %v", err) |
| 60 | } |
| 61 | front := httptest.NewServer(rt) |
| 62 | t.Cleanup(front.Close) |
| 63 | |
| 64 | req, _ := http.NewRequest("GET", front.URL+"/", nil) |
| 65 | req.Host = "nokey.example.com" |
| 66 | resp, err := http.DefaultClient.Do(req) |
| 67 | if err != nil { |
| 68 | t.Fatalf("request: %v", err) |
| 69 | } |
| 70 | defer resp.Body.Close() |
| 71 | if resp.StatusCode != http.StatusServiceUnavailable { |
| 72 | t.Fatalf("status %d, want 503", resp.StatusCode) |
| 73 | } |
| 74 | if ra := resp.Header.Get("Retry-After"); ra != "30" { |
| 75 | t.Fatalf("Retry-After %q, want 30", ra) |
| 76 | } |
| 77 | if a.entryCount() != 0 { |
| 78 | t.Fatal("refused wake left an entry behind") |
| 79 | } |
| 80 | // Nothing past Acquire ran: no restore, no spawn. |
| 81 | b.mu.Lock() |
| 82 | defer b.mu.Unlock() |
| 83 | if b.restores != 0 || b.starts != 0 { |
| 84 | t.Fatalf("refused wake went on: restores=%d starts=%d", b.restores, b.starts) |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | // A worker respawn keeps its lease and calls StartWorker directly, so it |
| 89 | // bypasses Acquire. The config gate has to be asked again there, and a refusal |
| 90 | // has to stand the worker down (stop it, release the lease) rather than leave |
| 91 | // a childless entry holding the lease from a sibling that could run it. |
| 92 | func TestWorkerRespawnIsRefusedByTheConfigGateAndStandsDown(t *testing.T) { |
| 93 | t.Parallel() |
| 94 | for _, why := range []string{"crash", "max_run"} { |
| 95 | t.Run(why, func(t *testing.T) { |
| 96 | a, b, clk, _ := testActivator(t, time.Hour) |
| 97 | rt := alwaysOnWorkerRoute("bot.x.internal") |
| 98 | rt.MaxRunSeconds = 30 |
| 99 | if err := a.WakeWorker(rt); err != nil { |
| 100 | t.Fatalf("WakeWorker: %v", err) |
| 101 | } |
| 102 | if b.workerStarts() != 1 { |
| 103 | t.Fatalf("workerStarts = %d, want 1", b.workerStarts()) |
| 104 | } |
| 105 | first := b.lastWorker() |
| 106 | |
| 107 | // Secrets arrive after the worker started; this box cannot open them. |
| 108 | b.mu.Lock() |
| 109 | b.admitErr = errors.New("no key") |
| 110 | b.mu.Unlock() |
| 111 | |
| 112 | if why == "crash" { |
| 113 | first.kill() |
| 114 | } else { |
| 115 | clk.advance(31 * time.Second) |
| 116 | } |
| 117 | a.sweepOnce() |
| 118 | |
| 119 | waitFor(t, "the refused worker to stand down", func() bool { return a.entryCount() == 0 }) |
| 120 | if n := b.workerStarts(); n != 1 { |
| 121 | t.Fatalf("workerStarts = %d: a refused respawn still started a child", n) |
| 122 | } |
| 123 | if !first.isStopped() && why == "max_run" { |
| 124 | t.Error("the old child was left running") |
| 125 | } |
| 126 | if n := b.releaseCount(); n != 1 { |
| 127 | t.Fatalf("lease releases = %d, want 1: a refused worker must give its lease back", n) |
| 128 | } |
| 129 | if !b.lastRep().isClosed() { |
| 130 | t.Error("replication was left open on a refused worker") |
| 131 | } |
| 132 | }) |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | // Build convergence shares the respawn helper but is its own trigger: a ship |
| 137 | // repoints live/<host>, the drift outlasts ConvergeGrace, and the always-on |
| 138 | // worker would be cycled onto the new build. With the gate refusing, it stands |
| 139 | // down instead of starting the new build without its config. |
| 140 | func TestWorkerConvergeRespawnIsRefusedByTheConfigGateAndStandsDown(t *testing.T) { |
| 141 | t.Parallel() |
| 142 | a, b, clk, ver := shippingActivator(t, time.Hour) |
| 143 | host := "bot.x.internal" |
| 144 | if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| 145 | t.Fatalf("WakeWorker: %v", err) |
| 146 | } |
| 147 | first := b.lastWorker() |
| 148 | b.mu.Lock() |
| 149 | b.admitErr = errors.New("cannot open sealed config") |
| 150 | b.mu.Unlock() |
| 151 | |
| 152 | ver.Store("v2") |
| 153 | a.sweepOnce() // first sighting of the drift |
| 154 | clk.advance(ConvergeGrace + time.Minute) |
| 155 | a.sweepOnce() // the converge respawn, refused |
| 156 | |
| 157 | waitFor(t, "the refused convergence to stand down", func() bool { return a.entryCount() == 0 }) |
| 158 | if n := b.workerStarts(); n != 1 { |
| 159 | t.Fatalf("workerStarts = %d: a refused convergence still started the new build", n) |
| 160 | } |
| 161 | if !first.isStopped() { |
| 162 | t.Error("the old child was left running") |
| 163 | } |
| 164 | if n := b.releaseCount(); n != 1 { |
| 165 | t.Fatalf("lease releases = %d, want 1", n) |
| 166 | } |
| 167 | if !b.lastRep().isClosed() { |
| 168 | t.Error("replication was left open") |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | // And a gate that allows it changes nothing: the respawn runs in place and |
| 173 | // keeps its lease, as before. |
| 174 | func TestWorkerRespawnProceedsWhenTheConfigGateAllows(t *testing.T) { |
| 175 | t.Parallel() |
| 176 | a, b, _, _ := testActivator(t, time.Hour) |
| 177 | if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| 178 | t.Fatalf("WakeWorker: %v", err) |
| 179 | } |
| 180 | b.lastWorker().kill() |
| 181 | a.sweepOnce() |
| 182 | waitForRestart(t, a, b, "bot.x.internal", 2) |
| 183 | if n := b.releaseCount(); n != 0 { |
| 184 | t.Fatalf("lease releases = %d, want 0", n) |
| 185 | } |
| 186 | b.mu.Lock() |
| 187 | admits := b.admits |
| 188 | b.mu.Unlock() |
| 189 | if admits == 0 { |
| 190 | t.Error("the respawn never asked the config gate") |
| 191 | } |
| 192 | } |
| 193 | |
| 194 | // The gate follows dispatch, which is Backing alone (SplitBackend.pick). A |
| 195 | // socket-backed hibernating instance may carry a Unit with an empty Backing |
| 196 | // and still reaches ExecBackend.Acquire, so it is gated; only Backing == unit |
| 197 | // is exempt. |
| 198 | func TestSplitBackendGatesWhatDispatchSendsToExec(t *testing.T) { |
| 199 | var asked []string |
| 200 | exec := &ExecBackend{ConfigGate: func(_ context.Context, rt registry.Route) error { |
| 201 | asked = append(asked, rt.Host) |
| 202 | return errors.New("no key") |
| 203 | }} |
| 204 | b := &SplitBackend{Exec: exec, Unit: &UnitBackend{}} |
| 205 | |
| 206 | unitNamed := registry.Route{Host: "named.example.com", Kind: "instance", App: "shop", |
| 207 | Socket: "/run/named.sock", Unit: "x.service", Hibernate: true} |
| 208 | if _, err := b.Acquire(context.Background(), unitNamed); err == nil { |
| 209 | t.Fatal("Acquire of a Unit-named, empty-Backing instance was not refused") |
| 210 | } |
| 211 | if err := b.AdmitConfig(context.Background(), unitNamed); err == nil { |
| 212 | t.Fatal("AdmitConfig of a Unit-named, empty-Backing instance was not refused") |
| 213 | } |
| 214 | unitBacked := unitNamed |
| 215 | unitBacked.Host, unitBacked.Backing = "backed.example.com", registry.BackingUnit |
| 216 | if err := b.AdmitConfig(context.Background(), unitBacked); err != nil { |
| 217 | t.Fatalf("a Backing=unit route must not be gated: %v", err) |
| 218 | } |
| 219 | if len(asked) != 2 || asked[0] != "named.example.com" || asked[1] != "named.example.com" { |
| 220 | t.Fatalf("gate asked %v, want only the Unit-named exec route, twice", asked) |
| 221 | } |
| 222 | } |
| 223 | |
amadan