| 1 | package relay |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "net" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "strconv" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | func TestReadPSI(t *testing.T) { |
| 20 | for _, tc := range []struct { |
| 21 | name string |
| 22 | body string |
| 23 | want float64 |
| 24 | err bool |
| 25 | }{ |
| 26 | { |
| 27 | name: "quiet box", |
| 28 | body: "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", |
| 29 | want: 0, |
| 30 | }, |
| 31 | { |
| 32 | // The `full` line is deliberately higher: taking it instead of |
| 33 | // `some` would read as pressure the relay must not shed on. |
| 34 | name: "under pressure", |
| 35 | body: "some avg10=23.47 avg60=11.20 avg300=3.01 total=91234567\nfull avg10=44.00 avg60=9.00 avg300=1.00 total=44\n", |
| 36 | want: 23.47, |
| 37 | }, |
| 38 | {name: "no some line", body: "full avg10=1.00 avg60=0 avg300=0 total=0\n", err: true}, |
| 39 | {name: "unparseable", body: "some avg10=lots avg60=0 avg300=0 total=0\n", err: true}, |
| 40 | {name: "empty", body: "", err: true}, |
| 41 | } { |
| 42 | t.Run(tc.name, func(t *testing.T) { |
| 43 | path := filepath.Join(t.TempDir(), "pressure") |
| 44 | if err := os.WriteFile(path, []byte(tc.body), 0o644); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | got, err := readPSI(path) |
| 48 | if tc.err { |
| 49 | if err == nil { |
| 50 | t.Fatalf("readPSI(%q) = %v, want an error", tc.body, got) |
| 51 | } |
| 52 | return |
| 53 | } |
| 54 | if err != nil { |
| 55 | t.Fatalf("readPSI: %v", err) |
| 56 | } |
| 57 | if got != tc.want { |
| 58 | t.Fatalf("readPSI = %v, want %v", got, tc.want) |
| 59 | } |
| 60 | }) |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | // A missing pressure file is the ordinary state on a kernel without PSI and |
| 65 | // on the platforms this binary cross-compiles for: an error the caller skips |
| 66 | // a tick on, never a panic. |
| 67 | func TestReadPSIMissingFile(t *testing.T) { |
| 68 | if _, err := readPSI(filepath.Join(t.TempDir(), "nope")); !errors.Is(err, os.ErrNotExist) { |
| 69 | t.Fatalf("readPSI on a missing file = %v, want ErrNotExist", err) |
| 70 | } |
| 71 | } |
| 72 | |
| 73 | func TestConfigFromEnvRefusesNoPublicIP(t *testing.T) { |
| 74 | _, err := ConfigFromEnv(context.Background(), func(string) string { return "" }, |
| 75 | func() string { return "" }, func() string { return "eu-west-1" }, newMemStore(t)) |
| 76 | if err == nil { |
| 77 | t.Fatal("ConfigFromEnv with no public IP anywhere: want an error") |
| 78 | } |
| 79 | if !strings.Contains(err.Error(), "CARLOS_RELAY_PUBLIC_IP") { |
| 80 | t.Fatalf("error should name the override: %v", err) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | func TestConfigFromEnv(t *testing.T) { |
| 85 | env := map[string]string{ |
| 86 | "CARLOS_RELAY_PUBLIC_IP": " 203.0.113.7 ", |
| 87 | "CARLOS_AWS_REGION": "us-east-1", |
| 88 | "CARLOS_RELAY_DENY_CIDRS": "10.4.0.0/16, 198.51.100.9/32", |
| 89 | "CARLOS_RELAY_MASTERS_PATH": "/tmp/masters", |
| 90 | "CARLOS_RELAY_BOX": "box-7", |
| 91 | "CARLOS_RELAY_LIMIT_APP_LIVE": "9", |
| 92 | "CARLOS_RELAY_PRESSURE_PATH": "/tmp/pressure", |
| 93 | "CARLOS_DEPLOYMENT_BUCKET_ETC": "ignored", |
| 94 | } |
| 95 | cfg, err := ConfigFromEnv(context.Background(), func(k string) string { return env[k] }, |
| 96 | func() string { return "192.0.2.1" }, func() string { return "eu-west-1" }, newMemStore(t)) |
| 97 | if err != nil { |
| 98 | t.Fatalf("ConfigFromEnv: %v", err) |
| 99 | } |
| 100 | // Both overrides must beat IMDS, and the region one must match |
| 101 | // edgesFileRegion's rule so the relay and the edges file agree. |
| 102 | if got := cfg.PublicIP.String(); got != "203.0.113.7" { |
| 103 | t.Errorf("PublicIP = %s, want the override", got) |
| 104 | } |
| 105 | if cfg.Region != "us-east-1" { |
| 106 | t.Errorf("Region = %s, want the override", cfg.Region) |
| 107 | } |
| 108 | if cfg.Box != "box-7" || cfg.MastersPath != "/tmp/masters" || cfg.PressurePath != "/tmp/pressure" { |
| 109 | t.Errorf("config = %+v", cfg) |
| 110 | } |
| 111 | if len(cfg.DenyCIDRs) != 2 || cfg.DenyCIDRs[0].String() != "10.4.0.0/16" || cfg.DenyCIDRs[1].String() != "198.51.100.9/32" { |
| 112 | t.Errorf("DenyCIDRs = %v", cfg.DenyCIDRs) |
| 113 | } |
| 114 | if cfg.Limits.AppLive != 9 { |
| 115 | t.Errorf("AppLive = %d, want the env limit", cfg.Limits.AppLive) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | // A malformed deny list is an operator typo that would otherwise open the |
| 120 | // VPC to anything that can reach 3478, so it stops the relay. |
| 121 | func TestConfigFromEnvRefusesBadDenyCIDR(t *testing.T) { |
| 122 | env := map[string]string{"CARLOS_RELAY_PUBLIC_IP": "203.0.113.7", "CARLOS_RELAY_DENY_CIDRS": "10.4.0.0/16,not-a-cidr"} |
| 123 | if _, err := ConfigFromEnv(context.Background(), func(k string) string { return env[k] }, nil, nil, newMemStore(t)); err == nil { |
| 124 | t.Fatal("want an error for a malformed CIDR") |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | // With no override, IMDS supplies both, and an off-EC2 box with neither |
| 129 | // region source is region-"" rather than an error: it serves STUN. |
| 130 | func TestConfigFromEnvFallsBackToIMDS(t *testing.T) { |
| 131 | cfg, err := ConfigFromEnv(context.Background(), func(string) string { return "" }, |
| 132 | func() string { return "192.0.2.5" }, func() string { return "" }, newMemStore(t)) |
| 133 | if err != nil { |
| 134 | t.Fatalf("ConfigFromEnv: %v", err) |
| 135 | } |
| 136 | if cfg.PublicIP.String() != "192.0.2.5" || cfg.Region != "" || cfg.MastersPath != defaultMastersPath { |
| 137 | t.Fatalf("config = %+v", cfg) |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | func TestKidsOfNilMasters(t *testing.T) { |
| 142 | if got := kidsOf(nil); got != nil { |
| 143 | t.Fatalf("kidsOf(nil) = %v, want nil", got) |
| 144 | } |
| 145 | m, err := ParseMasters(strings.NewReader("k1 " + b64(testMaster()) + " -\nk2 " + b64(testMaster()) + " -\n")) |
| 146 | if err != nil { |
| 147 | t.Fatal(err) |
| 148 | } |
| 149 | if got := kidsOf(m); len(got) != 2 || got[0] != "k1" || got[1] != "k2" { |
| 150 | t.Fatalf("kidsOf = %v", got) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // seedWindows must charge a restart's rebuilt window with in+out, keyed on |
| 155 | // the FIRST slash: an app name can never contain one, an account sqid never |
| 156 | // can either, and a malformed key is skipped rather than seeding a garbage |
| 157 | // identity. |
| 158 | func TestSeedWindows(t *testing.T) { |
| 159 | srv, _, _, _, _ := newTestServer(t, "203.0.113.7", nil, func(cfg *ServerConfig) { |
| 160 | cfg.Limits = Limits{AppLive: 1, AppPerMinute: 1, AppBytesPerHour: 1000, |
| 161 | AccountLive: 1, AccountPerMinute: 1, AccountBytesPerHour: 1 << 40, BoxLive: 1, BoxPerMinute: 1} |
| 162 | }) |
| 163 | at := time.Now().Add(-30 * time.Minute) |
| 164 | seedWindows(srv, []MeterSample{{ |
| 165 | At: at, |
| 166 | Apps: map[string]MeterApp{ |
| 167 | "abc/meet": {In: 400, Out: 300}, |
| 168 | "abc/other": {In: 1, Out: 1}, |
| 169 | "nokey": {In: 999999}, |
| 170 | }, |
| 171 | }}) |
| 172 | meet := Identity{Account: "abc", App: "meet"} |
| 173 | if srv.adm.OverBudget(meet, time.Now()) { |
| 174 | t.Fatal("700 of 1000 bytes seeded should not be over budget") |
| 175 | } |
| 176 | seedWindows(srv, []MeterSample{{At: at, Apps: map[string]MeterApp{"abc/meet": {In: 400}}}}) |
| 177 | if !srv.adm.OverBudget(meet, time.Now()) { |
| 178 | t.Fatal("1100 of 1000 bytes seeded should be over budget") |
| 179 | } |
| 180 | if srv.adm.OverBudget(Identity{Account: "nokey"}, time.Now()) { |
| 181 | t.Fatal("a key with no slash must not seed anything") |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | // The serve loop, end to end on ephemeral ports: it binds, reads the |
| 186 | // authority record, writes the heartbeat the console reads, and shuts down |
| 187 | // cleanly when its context is cancelled (systemd's SIGTERM). |
| 188 | func TestServeWritesHeartbeatAndStops(t *testing.T) { |
| 189 | store := newMemStore(t) |
| 190 | dir := t.TempDir() |
| 191 | masters := filepath.Join(dir, "relay-masters") |
| 192 | // k2 carries a retirement date, so this fixture also covers the hop that |
| 193 | // puts it on the heartbeat for `carlos relay status` (discussion #120). |
| 194 | if err := os.WriteFile(masters, []byte( |
| 195 | "k1 "+b64(testMaster())+" -\n"+ |
| 196 | "k2 "+b64(testMaster())+" 2026-09-09T12:00:00Z\n"), 0o640); err != nil { |
| 197 | t.Fatal(err) |
| 198 | } |
| 199 | cfg := Config{ |
| 200 | Box: "box-serve", |
| 201 | Region: "eu-west-1", |
| 202 | MastersPath: masters, |
| 203 | PublicIP: net.ParseIP("203.0.113.7"), |
| 204 | Store: store, |
| 205 | Limits: DefaultLimits(), |
| 206 | PressurePath: filepath.Join(dir, "no-pressure-here"), |
| 207 | udp: "127.0.0.1:0", |
| 208 | tcp: "127.0.0.1:0", |
| 209 | } |
| 210 | // The masters watcher outlives Serve by up to one poll interval, and |
| 211 | // t.Logf from a finished test is a data race, so the log goes quiet |
| 212 | // when the test body does. |
| 213 | var logMu sync.Mutex |
| 214 | quiet := false |
| 215 | logf := func(format string, args ...any) { |
| 216 | logMu.Lock() |
| 217 | defer logMu.Unlock() |
| 218 | if !quiet { |
| 219 | t.Logf(format, args...) |
| 220 | } |
| 221 | } |
| 222 | t.Cleanup(func() { logMu.Lock(); quiet = true; logMu.Unlock() }) |
| 223 | |
| 224 | ctx, cancel := context.WithCancel(context.Background()) |
| 225 | done := make(chan error, 1) |
| 226 | go func() { done <- Serve(ctx, cfg, logf) }() |
| 227 | |
| 228 | hb := waitHeartbeat(t, store, "box-serve") |
| 229 | if hb.Region != "eu-west-1" || len(hb.Listeners) != 2 { |
| 230 | t.Errorf("heartbeat = %+v, want both listeners and the region", hb) |
| 231 | } |
| 232 | if hb.ReadAt.IsZero() { |
| 233 | t.Error("heartbeat carries no read time, so the lease clock is invisible") |
| 234 | } |
| 235 | // Without this the relay half of the retiring display can be deleted with |
| 236 | // the whole suite green, and an operator mid-rotation silently reads |
| 237 | // "nothing is retiring". Polled rather than asserted on the heartbeat |
| 238 | // above: the masters watcher is a goroutine, so the FIRST heartbeat can |
| 239 | // legitimately be written before the file has been read. |
| 240 | deadline := time.Now().Add(10 * time.Second) |
| 241 | for { |
| 242 | hb := waitHeartbeat(t, store, "box-serve") |
| 243 | if got, ok := hb.Retiring["k2"]; ok { |
| 244 | if want := "2026-09-09T12:00:00Z"; got.UTC().Format(time.RFC3339) != want { |
| 245 | t.Errorf("k2 retire_at = %s, want %s", got.UTC().Format(time.RFC3339), want) |
| 246 | } |
| 247 | if _, ok := hb.Retiring["k1"]; ok { |
| 248 | t.Error("k1 has no retirement date and must not be reported as retiring") |
| 249 | } |
| 250 | break |
| 251 | } |
| 252 | if time.Now().After(deadline) { |
| 253 | t.Fatalf("no heartbeat ever carried k2's retirement (last: %+v)", hb.Retiring) |
| 254 | } |
| 255 | time.Sleep(50 * time.Millisecond) |
| 256 | } |
| 257 | |
| 258 | cancel() |
| 259 | select { |
| 260 | case err := <-done: |
| 261 | if err != nil { |
| 262 | t.Fatalf("Serve returned %v, want nil on a cancelled context", err) |
| 263 | } |
| 264 | case <-time.After(10 * time.Second): |
| 265 | t.Fatal("Serve did not return after its context was cancelled") |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | // A failed authority read must not look like a healthy one. read_at is the |
| 270 | // lease clock — the last SUCCESSFUL read — so the gap between it and the |
| 271 | // heartbeat's arrival is the relay's real drift from the record it serves; |
| 272 | // a wall clock there would say "just now" for a relay whose every read has |
| 273 | // failed for twenty minutes, and the console would believe it. |
| 274 | func TestServeHeartbeatIsHonestAboutAFailedRead(t *testing.T) { |
| 275 | inner := newMemStore(t) |
| 276 | store := &flakyStore{Store: inner} |
| 277 | dir := t.TempDir() |
| 278 | masters := filepath.Join(dir, "relay-masters") |
| 279 | if err := os.WriteFile(masters, []byte("k1 "+b64(testMaster())+" -\n"), 0o640); err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | cfg := Config{ |
| 283 | Box: "box-honest", Region: "eu-west-1", MastersPath: masters, |
| 284 | PublicIP: net.ParseIP("203.0.113.7"), Store: store, Limits: DefaultLimits(), |
| 285 | PressurePath: filepath.Join(dir, "none"), udp: "127.0.0.1:0", tcp: "127.0.0.1:0", |
| 286 | } |
| 287 | |
| 288 | var logMu sync.Mutex |
| 289 | quiet := false |
| 290 | logf := func(format string, args ...any) { |
| 291 | logMu.Lock() |
| 292 | defer logMu.Unlock() |
| 293 | if !quiet { |
| 294 | t.Logf(format, args...) |
| 295 | } |
| 296 | } |
| 297 | t.Cleanup(func() { logMu.Lock(); quiet = true; logMu.Unlock() }) |
| 298 | |
| 299 | ctx, cancel := context.WithCancel(context.Background()) |
| 300 | done := make(chan error, 1) |
| 301 | go func() { done <- Serve(ctx, cfg, logf) }() |
| 302 | |
| 303 | // The first read succeeds, so the lease clock is set and honest. |
| 304 | good := waitHeartbeat(t, inner, "box-honest") |
| 305 | if good.ReadAt.IsZero() || good.AuthorityError != "" { |
| 306 | t.Fatalf("first heartbeat = %+v, want a read time and no error", good) |
| 307 | } |
| 308 | cancel() |
| 309 | if err := <-done; err != nil { |
| 310 | t.Fatalf("Serve: %v", err) |
| 311 | } |
| 312 | |
| 313 | // Now every authority read fails. Serve is stopped, so drive readAuth's |
| 314 | // contract directly through a second run whose store is already broken: |
| 315 | // the lease clock must not move and the reason must be recorded. |
| 316 | store.failGet(AuthorityKey) |
| 317 | ctx2, cancel2 := context.WithCancel(context.Background()) |
| 318 | done2 := make(chan error, 1) |
| 319 | go func() { done2 <- Serve(ctx2, cfg, logf) }() |
| 320 | deadline := time.Now().Add(10 * time.Second) |
| 321 | var hb Heartbeat |
| 322 | for { |
| 323 | hb = waitHeartbeat(t, inner, "box-honest") |
| 324 | if hb.AuthorityError != "" || time.Now().After(deadline) { |
| 325 | break |
| 326 | } |
| 327 | time.Sleep(20 * time.Millisecond) |
| 328 | } |
| 329 | cancel2() |
| 330 | <-done2 |
| 331 | |
| 332 | if hb.AuthorityError == "" { |
| 333 | t.Fatalf("heartbeat = %+v, want the read error recorded", hb) |
| 334 | } |
| 335 | // A relay that has never read successfully has no lease and no read |
| 336 | // time to report — the one thing it must not do is invent one. |
| 337 | if !hb.ReadAt.IsZero() { |
| 338 | t.Errorf("read_at = %v on a relay whose every read failed, want zero", hb.ReadAt) |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | // flakyStore is a release.Store that can be told to fail Get for one key, |
| 343 | // so a test can watch what the relay says when the bucket stops answering. |
| 344 | type flakyStore struct { |
| 345 | release.Store |
| 346 | mu sync.Mutex |
| 347 | fail map[string]bool |
| 348 | } |
| 349 | |
| 350 | func (f *flakyStore) failGet(key string) { |
| 351 | f.mu.Lock() |
| 352 | if f.fail == nil { |
| 353 | f.fail = map[string]bool{} |
| 354 | } |
| 355 | f.fail[key] = true |
| 356 | f.mu.Unlock() |
| 357 | } |
| 358 | |
| 359 | func (f *flakyStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 360 | f.mu.Lock() |
| 361 | bad := f.fail[key] |
| 362 | f.mu.Unlock() |
| 363 | if bad { |
| 364 | return nil, "", errors.New("bucket unreachable") |
| 365 | } |
| 366 | return f.Store.Get(ctx, key) |
| 367 | } |
| 368 | |
| 369 | // OpenStoreWithRetry waits a store out rather than dying on it: the unit is |
| 370 | // Restart=always, systemd's start limit would park it in `failed` within |
| 371 | // ten seconds, and STUN — which needs no store at all — would go down with |
| 372 | // it. |
| 373 | func TestOpenStoreWithRetry(t *testing.T) { |
| 374 | defer swapRetryTimings(t)() |
| 375 | want := newMemStore(t) |
| 376 | calls := 0 |
| 377 | got, err := OpenStoreWithRetry(context.Background(), func(context.Context) (release.Store, error) { |
| 378 | calls++ |
| 379 | if calls < 3 { |
| 380 | return nil, errors.New("no credentials yet") |
| 381 | } |
| 382 | return want, nil |
| 383 | }, t.Logf) |
| 384 | if err != nil { |
| 385 | t.Fatalf("OpenStoreWithRetry: %v", err) |
| 386 | } |
| 387 | if calls != 3 { |
| 388 | t.Errorf("opener called %d times, want 3 (two failures then a success)", calls) |
| 389 | } |
| 390 | if got != want { |
| 391 | t.Errorf("returned a different store than the opener gave") |
| 392 | } |
| 393 | } |
| 394 | |
| 395 | // And it gives up the moment systemd asks the process to stop, rather than |
| 396 | // holding a shutdown open for the length of the backoff. |
| 397 | func TestOpenStoreWithRetryStopsOnCancel(t *testing.T) { |
| 398 | defer swapRetryTimings(t)() |
| 399 | ctx, cancel := context.WithCancel(context.Background()) |
| 400 | calls := 0 |
| 401 | _, err := OpenStoreWithRetry(ctx, func(context.Context) (release.Store, error) { |
| 402 | calls++ |
| 403 | if calls == 2 { |
| 404 | cancel() |
| 405 | } |
| 406 | return nil, errors.New("still down") |
| 407 | }, nil) |
| 408 | if !errors.Is(err, context.Canceled) { |
| 409 | t.Fatalf("err = %v, want context.Canceled", err) |
| 410 | } |
| 411 | } |
| 412 | |
| 413 | func swapRetryTimings(t *testing.T) func() { |
| 414 | t.Helper() |
| 415 | oi, om, ol := storeRetryInitial, storeRetryMax, storeRetryLogEvery |
| 416 | storeRetryInitial, storeRetryMax, storeRetryLogEvery = time.Millisecond, 2*time.Millisecond, time.Millisecond |
| 417 | return func() { storeRetryInitial, storeRetryMax, storeRetryLogEvery = oi, om, ol } |
| 418 | } |
| 419 | |
| 420 | // Serve refuses to start without the two things it cannot invent: the box |
| 421 | // label its objects are keyed by, and a store to write them to. |
| 422 | func TestServeRefusesIncompleteConfig(t *testing.T) { |
| 423 | base := Config{Box: "b", Store: newMemStore(t), PublicIP: net.ParseIP("203.0.113.7")} |
| 424 | for name, mutate := range map[string]func(*Config){ |
| 425 | "no box": func(c *Config) { c.Box = "" }, |
| 426 | "no store": func(c *Config) { c.Store = nil }, |
| 427 | "no ip": func(c *Config) { c.PublicIP = nil }, |
| 428 | } { |
| 429 | cfg := base |
| 430 | mutate(&cfg) |
| 431 | if err := Serve(context.Background(), cfg, nil); err == nil { |
| 432 | t.Errorf("%s: Serve returned nil, want an error", name) |
| 433 | } |
| 434 | } |
| 435 | } |
| 436 | |
| 437 | func waitHeartbeat(t *testing.T, store release.Store, box string) Heartbeat { |
| 438 | t.Helper() |
| 439 | deadline := time.Now().Add(10 * time.Second) |
| 440 | for { |
| 441 | body, _, err := store.Get(context.Background(), HeartbeatKey(box)) |
| 442 | if err == nil { |
| 443 | var hb Heartbeat |
| 444 | if err := json.Unmarshal(body, &hb); err != nil { |
| 445 | t.Fatalf("heartbeat is not JSON: %v", err) |
| 446 | } |
| 447 | return hb |
| 448 | } |
| 449 | if time.Now().After(deadline) { |
| 450 | t.Fatalf("no heartbeat at %s within the deadline (last error %v)", HeartbeatKey(box), err) |
| 451 | } |
| 452 | time.Sleep(20 * time.Millisecond) |
| 453 | } |
| 454 | } |
| 455 | |
| 456 | // heartbeatUDPPort pulls the udp listener's port out of a heartbeat, which is |
| 457 | // the only place a test outside Serve can learn the port it bound. |
| 458 | func heartbeatUDPPort(t *testing.T, hb Heartbeat) int { |
| 459 | t.Helper() |
| 460 | for _, l := range hb.Listeners { |
| 461 | if p, ok := strings.CutPrefix(l, "udp:"); ok { |
| 462 | n, err := strconv.Atoi(p) |
| 463 | if err != nil { |
| 464 | t.Fatalf("udp listener %q has no port: %v", l, err) |
| 465 | } |
| 466 | return n |
| 467 | } |
| 468 | } |
| 469 | t.Fatalf("heartbeat lists no udp listener: %v", hb.Listeners) |
| 470 | return 0 |
| 471 | } |
| 472 | |
| 473 | // TestServeEndsAllocationsOnDisable is the end-to-end disable path through |
| 474 | // the real Serve loop, which spec §7 asks for and the branch never carried |
| 475 | // (discussion #120): a pion client holding a live allocation, an app removed |
| 476 | // from the authority record, and the relay acting on it within one authority |
| 477 | // read rather than on a direct EndWhere call. |
| 478 | // |
| 479 | // The serial is what makes it a test and not a minute-long wait: the relay |
| 480 | // polls control/serial every three seconds and re-reads the authority record |
| 481 | // when it moves, which is the same path a console-mediated disable takes on |
| 482 | // a real box. |
| 483 | func TestServeEndsAllocationsOnDisable(t *testing.T) { |
| 484 | store := newMemStore(t) |
| 485 | dir := t.TempDir() |
| 486 | mastersPath := filepath.Join(dir, "relay-masters") |
| 487 | if err := os.WriteFile(mastersPath, []byte("k1 "+b64(testMaster())+" -\n"), 0o640); err != nil { |
| 488 | t.Fatal(err) |
| 489 | } |
| 490 | ctx := context.Background() |
| 491 | putAuthority := func(apps []AuthorityApp) time.Time { |
| 492 | t.Helper() |
| 493 | writtenAt := time.Now().UTC() |
| 494 | body, err := json.Marshal(Authority{Apps: apps, WrittenAt: writtenAt}) |
| 495 | if err != nil { |
| 496 | t.Fatal(err) |
| 497 | } |
| 498 | _, etag, _ := store.Get(ctx, AuthorityKey) |
| 499 | if _, err := store.Put(ctx, AuthorityKey, body, etag); err != nil { |
| 500 | t.Fatal(err) |
| 501 | } |
| 502 | if err := release.BumpSerial(ctx, store, time.Now()); err != nil { |
| 503 | t.Fatal(err) |
| 504 | } |
| 505 | return writtenAt |
| 506 | } |
| 507 | putAuthority([]AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}}) |
| 508 | |
| 509 | cfg := Config{ |
| 510 | Box: "box-disable", Region: "eu-west-1", MastersPath: mastersPath, |
| 511 | // The advertised relay address is deliberately NOT loopback, as on a |
| 512 | // real box: the peers are on loopback and a relay IP equal to them |
| 513 | // does not forward (TestServerRelaysBothLegs makes the same choice). |
| 514 | PublicIP: net.ParseIP("203.0.113.7"), Store: store, Limits: DefaultLimits(), |
| 515 | PressurePath: filepath.Join(dir, "no-pressure-here"), |
| 516 | udp: "127.0.0.1:0", |
| 517 | tcp: "127.0.0.1:0", |
| 518 | extraAllow: mustCIDRs("127.0.0.0/8"), |
| 519 | } |
| 520 | var logMu sync.Mutex |
| 521 | quiet := false |
| 522 | logf := func(format string, args ...any) { |
| 523 | logMu.Lock() |
| 524 | defer logMu.Unlock() |
| 525 | if !quiet { |
| 526 | t.Logf(format, args...) |
| 527 | } |
| 528 | } |
| 529 | t.Cleanup(func() { logMu.Lock(); quiet = true; logMu.Unlock() }) |
| 530 | |
| 531 | runCtx, cancel := context.WithCancel(ctx) |
| 532 | defer cancel() |
| 533 | done := make(chan error, 1) |
| 534 | go func() { done <- Serve(runCtx, cfg, logf) }() |
| 535 | |
| 536 | port := heartbeatUDPPort(t, waitHeartbeat(t, store, "box-disable")) |
| 537 | username, password := creds(time.Now(), "abc", "meet", "0123456789abcdef") |
| 538 | client := turnClient(t, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, username, password) |
| 539 | relayConn, err := client.Allocate() |
| 540 | if err != nil { |
| 541 | t.Fatalf("allocate against a relay that has read an enabling record: %v", err) |
| 542 | } |
| 543 | defer relayConn.Close() |
| 544 | |
| 545 | // A peer on loopback, permitted through the extraAllow seam, and one |
| 546 | // packet through the relay to prove the allocation carries traffic while |
| 547 | // the app IS enabled. Without this the "no traffic after" check below |
| 548 | // would pass against a relay that never forwarded anything at all. |
| 549 | peer, err := net.ListenPacket("udp4", "127.0.0.1:0") |
| 550 | if err != nil { |
| 551 | t.Fatal(err) |
| 552 | } |
| 553 | defer peer.Close() |
| 554 | peerAddr := peer.LocalAddr().(*net.UDPAddr) |
| 555 | if err := client.CreatePermission(peerAddr); err != nil { |
| 556 | t.Fatalf("permission for a loopback peer: %v", err) |
| 557 | } |
| 558 | if _, err := relayConn.WriteTo([]byte("before"), peerAddr); err != nil { |
| 559 | t.Fatalf("write through the relay: %v", err) |
| 560 | } |
| 561 | buf := make([]byte, 64) |
| 562 | peer.SetReadDeadline(time.Now().Add(5 * time.Second)) |
| 563 | if n, _, err := peer.ReadFrom(buf); err != nil || string(buf[:n]) != "before" { |
| 564 | t.Fatalf("the relay did not carry traffic while enabled: n=%d err=%v", n, err) |
| 565 | } |
| 566 | |
| 567 | // The disable: the same record without the app, and a serial bump so the |
| 568 | // relay re-reads inside three seconds rather than at the next minute. |
| 569 | disabledAt := putAuthority(nil) |
| 570 | |
| 571 | // Wait on the relay's OWN report that it is holding the new record — |
| 572 | // the heartbeat carries the WrittenAt of the authority copy it serves — |
| 573 | // rather than on an allocation failing. Retrying Allocate until any |
| 574 | // error appeared was the first shape of this test and it was not a test |
| 575 | // at all: a refusal for an unrelated reason (a per-app allocation cap, |
| 576 | // say) reads exactly like the disable, so the assertion passed with the |
| 577 | // app still enabled. |
| 578 | deadline := time.Now().Add(20 * time.Second) |
| 579 | for { |
| 580 | hb := waitHeartbeat(t, store, "box-disable") |
| 581 | if !hb.WrittenAt.Before(disabledAt) { |
| 582 | break |
| 583 | } |
| 584 | if time.Now().After(deadline) { |
| 585 | t.Fatalf("the relay never picked up the record written at %s (last heartbeat holds %s)", disabledAt, hb.WrittenAt) |
| 586 | } |
| 587 | time.Sleep(100 * time.Millisecond) |
| 588 | } |
| 589 | |
| 590 | // The allocation taken before the disable must stop carrying traffic. |
| 591 | // This is the assertion the test is named for, and the first version of |
| 592 | // it did not make: it only re-Allocated, which proves admission refuses |
| 593 | // a NEW allocation and says nothing about the live one. Deleting Serve's |
| 594 | // EndWhere(..., "authority") call left that version green. |
| 595 | if _, err := relayConn.WriteTo([]byte("after"), peerAddr); err == nil { |
| 596 | peer.SetReadDeadline(time.Now().Add(time.Second)) |
| 597 | if n, _, err := peer.ReadFrom(buf); err == nil { |
| 598 | t.Errorf("the relay forwarded %q after the app was disabled: the allocation did not end", buf[:n]) |
| 599 | } |
| 600 | } |
| 601 | |
| 602 | // And a credential that was good a moment ago must be refused, which is |
| 603 | // what a member sees after `relay disable`. |
| 604 | probe := turnClient(t, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, username, password) |
| 605 | if _, err := probe.Allocate(); err == nil { |
| 606 | t.Error("the relay allocated for an app the authority record no longer carries") |
| 607 | } |
| 608 | |
| 609 | cancel() |
| 610 | select { |
| 611 | case err := <-done: |
| 612 | if err != nil { |
| 613 | t.Fatalf("Serve returned %v, want nil on a cancelled context", err) |
| 614 | } |
| 615 | case <-time.After(10 * time.Second): |
| 616 | t.Fatal("Serve did not return after its context was cancelled") |
| 617 | } |
| 618 | } |
| 619 | |
amadan