| 1 | package instancectl |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "io" |
| 6 | "net" |
| 7 | "net/http" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | // Astra round 3 P1: connection 1 (keep-alive) is accepted first and later |
| 14 | // carries the /stop; a delayed /start arrives on connection 2 and has not |
| 15 | // registered when the stop fences the host. The fence is the accepted |
| 16 | // connection watermark at that moment, not the stop's own number, so the |
| 17 | // delayed start is refused when it finally registers. |
| 18 | func TestADelayedStartAcceptedBeforeTheFenceIsRefusedWhenItRegisters(t *testing.T) { |
| 19 | host := "watermark.example.com" |
| 20 | s, _ := newServer(t, &fakeCtl{}) |
| 21 | _ = s.nextConnSeq() // connection 1: will carry the /stop |
| 22 | startConn := s.nextConnSeq() // connection 2: the delayed /start |
| 23 | release, err := s.fenceLegacyStarts(context.Background(), host) |
| 24 | if err != nil { |
| 25 | t.Fatalf("fence: %v", err) |
| 26 | } |
| 27 | release() |
| 28 | if _, finish, err := s.beginLegacyStart(context.Background(), host, startConn); err == nil { |
| 29 | finish() |
| 30 | t.Fatal("a start accepted before the fence registered after it and was admitted") |
| 31 | } |
| 32 | } |
| 33 | |
| 34 | // Astra round 3 P2: a client that reuses its connection gets a fresh stamp |
| 35 | // for every request, because the server answers each on its own connection. |
| 36 | // A start after a stop, from the same keep-alive client, runs. |
| 37 | func TestAKeepAliveClientsStartAfterAStopRuns(t *testing.T) { |
| 38 | host := "reuse.example.com" |
| 39 | ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}} |
| 40 | _, sock := serveOnSocket(t, ctl) |
| 41 | var dials int |
| 42 | c := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 43 | dials++ |
| 44 | return (&net.Dialer{}).DialContext(ctx, "unix", sock) |
| 45 | }}} |
| 46 | for i, path := range []string{"/start?host=" + host + "&role=run&memory_max=536870912", "/stop?host=" + host + "&role=run", "/start?host=" + host + "&role=run&memory_max=536870912"} { |
| 47 | resp, err := c.Post("http://unix"+path, "", nil) |
| 48 | if err != nil { |
| 49 | t.Fatalf("POST %s: %v", path, err) |
| 50 | } |
| 51 | body, _ := io.ReadAll(resp.Body) |
| 52 | resp.Body.Close() |
| 53 | if resp.StatusCode/100 != 2 { |
| 54 | t.Fatalf("request %d %s: status %d: %s", i, path, resp.StatusCode, body) |
| 55 | } |
| 56 | } |
| 57 | if dials != 3 { |
| 58 | t.Fatalf("dials = %d, want one connection per request", dials) |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | // Liveness 1: the fence covers only starts accepted before the stop. A |
| 63 | // /start accepted after the /stop has returned runs normally, and nothing |
| 64 | // of a fence outlives the server: a restarted instance-control has none. |
| 65 | func TestTheFenceCoversOnlyStartsAcceptedBeforeTheStop(t *testing.T) { |
| 66 | host := "after.example.com" |
| 67 | ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}} |
| 68 | s, c := newServer(t, ctl) |
| 69 | stop := post(t, c, "/stop?host="+host+"&role=run") |
| 70 | stop.Body.Close() |
| 71 | if stop.StatusCode != http.StatusNoContent { |
| 72 | t.Fatalf("/stop status %d", stop.StatusCode) |
| 73 | } |
| 74 | start := post(t, c, "/start?host="+host+"&role=run&memory_max=536870912") |
| 75 | start.Body.Close() |
| 76 | if start.StatusCode != http.StatusOK { |
| 77 | t.Fatalf("a start accepted after the stop returned: status %d, want 200", start.StatusCode) |
| 78 | } |
| 79 | s.legacyMu.Lock() |
| 80 | stopping := s.legacyStopping[host] |
| 81 | s.legacyMu.Unlock() |
| 82 | if stopping != 0 { |
| 83 | t.Fatalf("the stop's refusal of new starts outlived it (%d)", stopping) |
| 84 | } |
| 85 | // A restarted service: the fence lived in memory only. |
| 86 | _, c2 := newServer(t, &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv2\n"}}) |
| 87 | again := post(t, c2, "/start?host="+host+"&role=run&memory_max=536870912") |
| 88 | again.Body.Close() |
| 89 | if again.StatusCode != http.StatusOK { |
| 90 | t.Fatalf("a restarted instance-control refused a start: status %d", again.StatusCode) |
| 91 | } |
| 92 | } |
| 93 | |
| 94 | // heldStart parks host's legacy /start before its tier step until release |
| 95 | // is closed, and returns once the handler is parked there. The handler does |
| 96 | // not see its own cancellation (unseenCancel). |
| 97 | func heldStart(t *testing.T, ctl *fakeCtl, c *http.Client, host string) (release chan struct{}) { |
| 98 | t.Helper() |
| 99 | entered, gate := make(chan struct{}), make(chan struct{}) |
| 100 | ctl.mu.Lock() |
| 101 | if ctl.before == nil { |
| 102 | ctl.before = map[string]func(){} |
| 103 | } |
| 104 | ctl.before["set-property"] = func() { |
| 105 | close(entered) |
| 106 | <-gate |
| 107 | } |
| 108 | ctl.unseenCancel = true |
| 109 | ctl.mu.Unlock() |
| 110 | postAsync(c, "/start?host="+host+"&role=run&memory_max=536870912") |
| 111 | select { |
| 112 | case <-entered: |
| 113 | case <-time.After(5 * time.Second): |
| 114 | t.Fatal("the held start never reached its tier step") |
| 115 | } |
| 116 | return gate |
| 117 | } |
| 118 | |
| 119 | // Liveness 2: the join is per host. Host A's attempt will not join; /start |
| 120 | // and /stop of host B answer at once while A's /stop is still joining. |
| 121 | func TestAStuckJoinDelaysNoOtherHost(t *testing.T) { |
| 122 | ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}} |
| 123 | _, c := newServer(t, ctl) |
| 124 | gate := heldStart(t, ctl, c, "stuck.example.com") |
| 125 | stopA := postAsync(c, "/stop?host=stuck.example.com&role=run") |
| 126 | defer func() { close(gate); <-stopA }() |
| 127 | time.Sleep(50 * time.Millisecond) // A's stop is joining |
| 128 | for _, path := range []string{"/start?host=free.example.com&role=run&memory_max=536870912", "/stop?host=free.example.com&role=run"} { |
| 129 | done := postAsync(c, path) |
| 130 | select { |
| 131 | case code := <-done: |
| 132 | if code/100 != 2 { |
| 133 | t.Fatalf("%s: status %d", path, code) |
| 134 | } |
| 135 | case <-time.After(2 * time.Second): |
| 136 | t.Fatalf("%s waited behind another host's stuck join", path) |
| 137 | } |
| 138 | } |
| 139 | select { |
| 140 | case <-stopA: |
| 141 | t.Fatal("A's stop answered while its attempt was still stuck") |
| 142 | default: |
| 143 | } |
| 144 | } |
| 145 | |
| 146 | // Liveness 3: a stuck stop answers once, within the bound, and the next |
| 147 | // /stop of the host does not hang behind the stuck attempt: it answers the |
| 148 | // same error within the bound, and joins it once it returns. |
| 149 | func TestAStuckStopAnswersAndTheNextStopDoesNotHang(t *testing.T) { |
| 150 | host := "wedged.example.com" |
| 151 | ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}} |
| 152 | _, c := newServer(t, ctl, func(s *Server) { s.legacyJoin = 200 * time.Millisecond }) |
| 153 | gate := heldStart(t, ctl, c, host) |
| 154 | stopOnce := func() (int, string, time.Duration) { |
| 155 | begun := time.Now() |
| 156 | resp := post(t, c, "/stop?host="+host+"&role=run") |
| 157 | body, _ := io.ReadAll(resp.Body) |
| 158 | resp.Body.Close() |
| 159 | return resp.StatusCode, string(body), time.Since(begun) |
| 160 | } |
| 161 | for i := 0; i < 2; i++ { |
| 162 | code, body, took := stopOnce() |
| 163 | if code != http.StatusBadGateway || !strings.Contains(body, "still in flight") { |
| 164 | t.Fatalf("stop %d with the attempt stuck: %d %q, want 502 naming the start in flight", i+1, code, body) |
| 165 | } |
| 166 | if took > 2*time.Second { |
| 167 | t.Fatalf("stop %d took %s, past its bound", i+1, took) |
| 168 | } |
| 169 | } |
| 170 | close(gate) |
| 171 | time.Sleep(50 * time.Millisecond) |
| 172 | if code, body, _ := stopOnce(); code != http.StatusNoContent { |
| 173 | t.Fatalf("stop once the attempt returned: %d %q, want 204", code, body) |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | // Liveness 4: the other verbs pass through unchanged during a legacy stop. |
| 178 | // While host A's legacy /stop is joining a stuck attempt, A's sidecar |
| 179 | // /start and seed /stop answer at once. |
| 180 | func TestOtherVerbsPassThroughDuringALegacyStop(t *testing.T) { |
| 181 | host := "verbs.example.com" |
| 182 | ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}} |
| 183 | _, c := newServer(t, ctl) |
| 184 | gate := heldStart(t, ctl, c, host) |
| 185 | stop := postAsync(c, "/stop?host="+host+"&role=run") |
| 186 | defer func() { close(gate); <-stop }() |
| 187 | time.Sleep(50 * time.Millisecond) // the legacy stop is joining |
| 188 | for _, path := range []string{"/start?host=" + host + "&role=sidecar&memory_max=536870912", "/stop?host=" + host + "&role=seed"} { |
| 189 | done := postAsync(c, path) |
| 190 | select { |
| 191 | case code := <-done: |
| 192 | if code/100 != 2 { |
| 193 | t.Fatalf("%s during a legacy stop: status %d", path, code) |
| 194 | } |
| 195 | case <-time.After(2 * time.Second): |
| 196 | t.Fatalf("%s waited behind a legacy stop", path) |
| 197 | } |
| 198 | } |
| 199 | } |
| 200 | |
| 201 | // postAsync POSTs path from its own goroutine, never touching t, and |
| 202 | // delivers the status (0 on a transport error). |
| 203 | func postAsync(c *http.Client, path string) <-chan int { |
| 204 | done := make(chan int, 1) |
| 205 | go func() { |
| 206 | resp, err := c.Post("http://unix"+path, "", nil) |
| 207 | if err != nil { |
| 208 | done <- 0 |
| 209 | return |
| 210 | } |
| 211 | resp.Body.Close() |
| 212 | done <- resp.StatusCode |
| 213 | }() |
| 214 | return done |
| 215 | } |
| 216 | |
amadan