| 1 | # Stage 4a relay (STUN and TURN) — Implementation Plan |
| 2 | |
| 3 | > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. |
| 4 | |
| 5 | **Goal:** A platform-run STUN and TURN relay on every armed region box, enabled per app through `carlos relay enable`, with keys delivered as sealed config, so the Oficina Meet app connects calls across the internet with no third party in the path. |
| 6 | |
| 7 | **Architecture:** One new package, `internal/relay`, holds everything pure: key derivation, the credential contract, the authority record, the pion server wrapper with its own socket wrapping for termination and metering, and the limits. The console gains a relay record per app, a relay-writer lease, an aggregate publisher and a reconcile, plus four CLI doors. The host gains a `carlos-relay.service` unit, a `carlos relay serve` verb, and a per-route config acknowledgement written by the config sync. Infra gains an `extra_ingress` block and a DNS record per region. |
| 8 | |
| 9 | **Tech Stack:** Go 1.26, `github.com/pion/turn/v4` v4.1.4 (new dependency, pure Go, cross-compiles clean; checked 2026-09-07), `modernc.org/sqlite` (existing), OpenTofu (infra). |
| 10 | |
| 11 | **Spec:** `docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md` (revision 6, codex accept). Parent: `docs/superpowers/specs/2026-09-07-media-platform-roadmap-design.md`. |
| 12 | |
| 13 | ## Global Constraints |
| 14 | |
| 15 | - **The app-facing contract (spec §2) is frozen.** `CARLOS_RELAY` is standard padded base64 of a JSON array of `{region, urls, key, username_suffix}`; username is `<expiry>:<account>:<app>:<incarnation>:<kid>`; credential is `base64_std(HMAC-SHA256(key, username))`; max expiry four hours ahead. Test vector: master `SHA-256("carlos-relay-test-master")`, account `abc`, app `meet`, incarnation `0123456789abcdef`, kid `k1`, expiry `1900000000` → key `uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=`, credential `jRU479Y3GO45nrRHqxBf3BMeBHE6Fip1cwAccNv28MQ=`. Any change is a spec change. |
| 16 | - **Rulings (spec §9) are fixed:** four-hour credentials; member-level enable; console holds masters as sealed config; no TLS listener; two meters (`in`, `out`); limits as in spec §3. |
| 17 | - **Wire shapes are additive.** Every new bucket object is a new key; `relay.json`, `control/relay.json`, `control/relay-issuer.json`, `control/relay-writer.json`, `control/relay-ack/<box>.json`, `control/config-ack/<box>/<host>.json`, `metrics/relay/<box>/<hour>.json`. Nothing existing is retyped. |
| 18 | - **CAS is `Put(ctx, key, body, ifMatch)`** on `release.Store`; a mismatch returns `release.ErrCASConflict`. There is no other conditional primitive. (`internal/release/store.go:42`.) |
| 19 | - **Config values must pass `appconfig.ValidVarValue`**: no quotes, backslashes, newlines or NUL. That is why `CARLOS_RELAY` is base64. |
| 20 | - **`CGO_ENABLED=0` everywhere.** The CLI cross-compiles for darwin/arm64, darwin/amd64, linux/arm64, windows/amd64, windows/arm64 (`.amadan/ci.d/20-cross@5`). pion/turn is pure Go. |
| 21 | - **Every `carlos` command needs a `### carlos <verb>` heading in `docs/site/cli.md`** or `internal/docsite` fails the gate. |
| 22 | - **Build gate before any commit lands:** `go build ./...`, `go vet ./...`, `go test ./... -count=1`, `gofmt -l .` clean. Sandbox: `GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off`; adding the dependency needs the sandbox lifted once for `go get`. |
| 23 | - **The forge is `origin` (amadan.net).** Push there, `amadan branch describe`, wait for `amadan ci status -branch` to pass, `amadan branch merge … -expect <full sha>`. Never `gh`. |
| 24 | - **Delegation (AGENTS.md):** Tasks 1 to 5 are Sonnet from this brief; Task 6 (pion wrapper) and Task 10 (console operations with the lease) are Opus; final whole-branch review is Opus. |
| 25 | |
| 26 | --- |
| 27 | |
| 28 | ## File structure |
| 29 | |
| 30 | | Path | Responsibility | |
| 31 | |---|---| |
| 32 | | `internal/relay/keys.go` | HKDF derivation, credential minting, username parse; the §2 contract | |
| 33 | | `internal/relay/masters.go` | masters file parse, `retire_at`, watched reload | |
| 34 | | `internal/relay/authority.go` | `control/relay.json` shape, read with lease clock, heartbeat write | |
| 35 | | `internal/relay/limits.go` | admission counters, rolling-hour budgets, token buckets | |
| 36 | | `internal/relay/conn.go` | wrapped `net.PacketConn` and listener conns: counting, buckets, EIP port gate, close | |
| 37 | | `internal/relay/server.go` | pion `ServerConfig` assembly: generator, auth handler, permission handler, bindings, termination timers | |
| 38 | | `internal/relay/meter.go` | per-app counters, minute samples, hourly object read-append-CAS | |
| 39 | | `internal/relay/serve.go` | `Serve(ctx, Config)`: wires the above, env parsing, PSI shedding loop | |
| 40 | | `internal/relay/record.go` | `apps/<acct>/<app>/relay.json` shape, Get/Put, transitions | |
| 41 | | `internal/relay/config.go` | build the `CARLOS_RELAY` value from a record and issuer masters | |
| 42 | | `internal/relay/vector_test.go` | the §2 test vector | |
| 43 | | `internal/host/units/carlos-relay.service` | the unit | |
| 44 | | `internal/host/bootstrap.go` | arm and start the unit | |
| 45 | | `internal/host/systemapply.go` | restart the relay on platform update | |
| 46 | | `internal/appconfig/sync.go`, `internal/appconfig/configack.go` | per-route config acknowledgement | |
| 47 | | `internal/console/apirelay.go` | doors: enable, status, disable, adopted, rotate | |
| 48 | | `internal/console/relayops.go` | writer lease, operations, aggregate publish, reconcile | |
| 49 | | `relay.go` (root) | `carlos relay` CLI: enable, status, disable, adopted, rotate, keygen, serve | |
| 50 | | `apimode.go` | `apiSession.relay*` client calls | |
| 51 | | `infra/modules/region-host/main.tf`, `variables.tf`, `outputs.tf` | `relay` variable, ingress, VPC CIDR output | |
| 52 | | `docs/site/cli.md` | `### carlos relay` | |
| 53 | |
| 54 | --- |
| 55 | |
| 56 | ### Task 1: Key derivation and the credential contract |
| 57 | |
| 58 | **Files:** |
| 59 | - Create: `internal/relay/keys.go` |
| 60 | - Test: `internal/relay/vector_test.go`, `internal/relay/keys_test.go` |
| 61 | |
| 62 | **Interfaces:** |
| 63 | - Produces: |
| 64 | ```go |
| 65 | const MaxCredentialLife = 4 * time.Hour |
| 66 | func DeriveKey(master []byte, account, app, incarnation string) []byte // 32 bytes |
| 67 | func Suffix(account, app, incarnation, kid string) string // "acct:app:inc:kid" |
| 68 | func Credential(key []byte, username string) string // base64 std |
| 69 | type Username struct{ Expiry time.Time; Account, App, Incarnation, Kid string } |
| 70 | func ParseUsername(s string) (Username, error) |
| 71 | ``` |
| 72 | |
| 73 | - [ ] **Step 1: Write the vector test** |
| 74 | |
| 75 | ```go |
| 76 | package relay |
| 77 | |
| 78 | import ( |
| 79 | "crypto/sha256" |
| 80 | "encoding/base64" |
| 81 | "testing" |
| 82 | ) |
| 83 | |
| 84 | func TestVector(t *testing.T) { |
| 85 | m := sha256.Sum256([]byte("carlos-relay-test-master")) |
| 86 | key := DeriveKey(m[:], "abc", "meet", "0123456789abcdef") |
| 87 | if got := base64.StdEncoding.EncodeToString(key); got != "uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=" { |
| 88 | t.Fatalf("key = %s", got) |
| 89 | } |
| 90 | u := "1900000000:" + Suffix("abc", "meet", "0123456789abcdef", "k1") |
| 91 | if u != "1900000000:abc:meet:0123456789abcdef:k1" { |
| 92 | t.Fatalf("username = %s", u) |
| 93 | } |
| 94 | if got := Credential(key, u); got != "jRU479Y3GO45nrRHqxBf3BMeBHE6Fip1cwAccNv28MQ=" { |
| 95 | t.Fatalf("credential = %s", got) |
| 96 | } |
| 97 | } |
| 98 | ``` |
| 99 | |
| 100 | - [ ] **Step 2: Write the parse tests** |
| 101 | |
| 102 | ```go |
| 103 | func TestParseUsername(t *testing.T) { |
| 104 | u, err := ParseUsername("1900000000:abc:meet:0123456789abcdef:k1") |
| 105 | if err != nil || u.Account != "abc" || u.App != "meet" || u.Incarnation != "0123456789abcdef" || u.Kid != "k1" || u.Expiry.Unix() != 1900000000 { |
| 106 | t.Fatalf("parse = %+v %v", u, err) |
| 107 | } |
| 108 | for _, bad := range []string{"", "1:2:3:4", "1:2:3:4:5:6", "x:abc:meet:0123456789abcdef:k1", |
| 109 | "1900000000:ABC:meet:0123456789abcdef:k1", "1900000000:abc:9meet:0123456789abcdef:k1", |
| 110 | "1900000000:abc:meet:short:k1", "1900000000:abc:meet:0123456789abcdef:K1"} { |
| 111 | if _, err := ParseUsername(bad); err == nil { |
| 112 | t.Errorf("ParseUsername(%q) accepted", bad) |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | ``` |
| 117 | |
| 118 | - [ ] **Step 3: Run tests to verify they fail** |
| 119 | |
| 120 | Run: `go test ./internal/relay/ -run 'TestVector|TestParseUsername' -v` |
| 121 | Expected: FAIL, package does not exist. |
| 122 | |
| 123 | - [ ] **Step 4: Implement** |
| 124 | |
| 125 | ```go |
| 126 | // Package relay is the platform STUN/TURN relay: the app-facing credential |
| 127 | // contract (spec §2), key derivation (§4), and the server (§3). |
| 128 | package relay |
| 129 | |
| 130 | import ( |
| 131 | "crypto/hmac" |
| 132 | "crypto/sha256" |
| 133 | "encoding/base64" |
| 134 | "fmt" |
| 135 | "regexp" |
| 136 | "strconv" |
| 137 | "strings" |
| 138 | "time" |
| 139 | |
| 140 | "golang.org/x/crypto/hkdf" |
| 141 | "io" |
| 142 | ) |
| 143 | |
| 144 | const ( |
| 145 | MaxCredentialLife = 4 * time.Hour |
| 146 | hkdfSalt = "carlos-relay-v1" |
| 147 | ) |
| 148 | |
| 149 | var ( |
| 150 | accountRe = regexp.MustCompile(`^[a-z0-9]+$`) |
| 151 | appRe = regexp.MustCompile(`^[a-z][a-z0-9-]{0,31}$`) |
| 152 | incarnationRe = regexp.MustCompile(`^[0-9a-f]{16}$`) |
| 153 | kidRe = regexp.MustCompile(`^[a-z0-9]{1,8}$`) |
| 154 | ) |
| 155 | |
| 156 | // DeriveKey is HKDF-SHA256(master, salt "carlos-relay-v1", |
| 157 | // info account||0x00||app||0x00||incarnation, 32). |
| 158 | func DeriveKey(master []byte, account, app, incarnation string) []byte { |
| 159 | info := []byte(account + "\x00" + app + "\x00" + incarnation) |
| 160 | r := hkdf.New(sha256.New, master, []byte(hkdfSalt), info) |
| 161 | key := make([]byte, 32) |
| 162 | if _, err := io.ReadFull(r, key); err != nil { |
| 163 | panic("relay: hkdf: " + err.Error()) // cannot fail for 32 bytes |
| 164 | } |
| 165 | return key |
| 166 | } |
| 167 | |
| 168 | func Suffix(account, app, incarnation, kid string) string { |
| 169 | return account + ":" + app + ":" + incarnation + ":" + kid |
| 170 | } |
| 171 | |
| 172 | // Credential is base64(HMAC-SHA256(key, username)), the TURN password. |
| 173 | func Credential(key []byte, username string) string { |
| 174 | h := hmac.New(sha256.New, key) |
| 175 | h.Write([]byte(username)) |
| 176 | return base64.StdEncoding.EncodeToString(h.Sum(nil)) |
| 177 | } |
| 178 | |
| 179 | type Username struct { |
| 180 | Expiry time.Time |
| 181 | Account, App, Incarnation, Kid string |
| 182 | } |
| 183 | |
| 184 | // ParseUsername splits "<expiry>:<account>:<app>:<incarnation>:<kid>" and |
| 185 | // validates each field's alphabet. It does not check expiry against now; |
| 186 | // the auth handler does that with its own clock. |
| 187 | func ParseUsername(s string) (Username, error) { |
| 188 | f := strings.Split(s, ":") |
| 189 | if len(f) != 5 { |
| 190 | return Username{}, fmt.Errorf("relay: username has %d fields, want 5", len(f)) |
| 191 | } |
| 192 | exp, err := strconv.ParseInt(f[0], 10, 64) |
| 193 | if err != nil || exp < 0 { |
| 194 | return Username{}, fmt.Errorf("relay: bad expiry %q", f[0]) |
| 195 | } |
| 196 | switch { |
| 197 | case !accountRe.MatchString(f[1]): |
| 198 | return Username{}, fmt.Errorf("relay: bad account %q", f[1]) |
| 199 | case !appRe.MatchString(f[2]): |
| 200 | return Username{}, fmt.Errorf("relay: bad app %q", f[2]) |
| 201 | case !incarnationRe.MatchString(f[3]): |
| 202 | return Username{}, fmt.Errorf("relay: bad incarnation %q", f[3]) |
| 203 | case !kidRe.MatchString(f[4]): |
| 204 | return Username{}, fmt.Errorf("relay: bad kid %q", f[4]) |
| 205 | } |
| 206 | return Username{Expiry: time.Unix(exp, 0).UTC(), Account: f[1], App: f[2], Incarnation: f[3], Kid: f[4]}, nil |
| 207 | } |
| 208 | ``` |
| 209 | |
| 210 | `golang.org/x/crypto` is already in `go.mod`; `hkdf` is a subpackage of it. |
| 211 | |
| 212 | - [ ] **Step 5: Run tests to verify they pass** |
| 213 | |
| 214 | Run: `go test ./internal/relay/ -v` |
| 215 | Expected: PASS. |
| 216 | |
| 217 | - [ ] **Step 6: Commit** |
| 218 | |
| 219 | ```bash |
| 220 | git add internal/relay/ |
| 221 | git commit -m "relay: key derivation and the credential contract (spec §2, §4)" |
| 222 | ``` |
| 223 | |
| 224 | --- |
| 225 | |
| 226 | ### Task 2: The masters file |
| 227 | |
| 228 | **Files:** |
| 229 | - Create: `internal/relay/masters.go` |
| 230 | - Test: `internal/relay/masters_test.go` |
| 231 | |
| 232 | **Interfaces:** |
| 233 | - Produces: |
| 234 | ```go |
| 235 | type Master struct{ Kid string; Key []byte; RetireAt time.Time } // zero RetireAt = never |
| 236 | type Masters struct{ /* unexported */ } |
| 237 | func ParseMasters(r io.Reader) (*Masters, error) |
| 238 | func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) // false if absent or retired |
| 239 | func (m *Masters) Kids() []string // loaded kids, sorted |
| 240 | func WatchMasters(ctx context.Context, path string, onChange func(*Masters, error)) |
| 241 | ``` |
| 242 | - File format, one master per line: `<kid> <base64 key> <retire_at RFC3339 or "-">`. |
| 243 | |
| 244 | - [ ] **Step 1: Write the tests** |
| 245 | |
| 246 | ```go |
| 247 | func TestParseMasters(t *testing.T) { |
| 248 | in := "k1 " + b64(bytes.Repeat([]byte{1}, 32)) + " -\n" + |
| 249 | "k0 " + b64(bytes.Repeat([]byte{2}, 32)) + " 2026-09-08T00:00:00Z\n" + |
| 250 | "# comment\n\n" |
| 251 | m, err := ParseMasters(strings.NewReader(in)) |
| 252 | if err != nil { |
| 253 | t.Fatal(err) |
| 254 | } |
| 255 | if got := m.Kids(); !reflect.DeepEqual(got, []string{"k0", "k1"}) { |
| 256 | t.Fatalf("kids = %v", got) |
| 257 | } |
| 258 | before := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC) |
| 259 | after := time.Date(2026, 9, 9, 0, 0, 0, 0, time.UTC) |
| 260 | if _, ok := m.Lookup("k0", before); !ok { |
| 261 | t.Fatal("k0 should be live before retire_at") |
| 262 | } |
| 263 | if _, ok := m.Lookup("k0", after); ok { |
| 264 | t.Fatal("k0 should be retired after retire_at") |
| 265 | } |
| 266 | if _, ok := m.Lookup("k1", after); !ok { |
| 267 | t.Fatal("k1 never retires") |
| 268 | } |
| 269 | if _, ok := m.Lookup("kx", before); ok { |
| 270 | t.Fatal("unknown kid") |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | func TestParseMastersRejectsBadLines(t *testing.T) { |
| 275 | for _, bad := range []string{"k1 notbase64 -", "k1 " + b64([]byte{1}) + " -", "K1 " + b64(bytes.Repeat([]byte{1}, 32)) + " -", "k1 " + b64(bytes.Repeat([]byte{1}, 32)) + " yesterday"} { |
| 276 | if _, err := ParseMasters(strings.NewReader(bad)); err == nil { |
| 277 | t.Errorf("accepted %q", bad) |
| 278 | } |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | func b64(b []byte) string { return base64.StdEncoding.EncodeToString(b) } |
| 283 | ``` |
| 284 | |
| 285 | - [ ] **Step 2: Run to verify failure** |
| 286 | |
| 287 | Run: `go test ./internal/relay/ -run TestParseMasters -v` |
| 288 | Expected: FAIL, undefined. |
| 289 | |
| 290 | - [ ] **Step 3: Implement** |
| 291 | |
| 292 | ```go |
| 293 | package relay |
| 294 | |
| 295 | import ( |
| 296 | "bufio" |
| 297 | "context" |
| 298 | "encoding/base64" |
| 299 | "fmt" |
| 300 | "io" |
| 301 | "os" |
| 302 | "sort" |
| 303 | "strings" |
| 304 | "time" |
| 305 | ) |
| 306 | |
| 307 | type Master struct { |
| 308 | Kid string |
| 309 | Key []byte |
| 310 | RetireAt time.Time // zero means never |
| 311 | } |
| 312 | |
| 313 | type Masters struct{ byKid map[string]Master } |
| 314 | |
| 315 | // ParseMasters reads "<kid> <base64 key> <retire_at|->" lines. Blank lines |
| 316 | // and "#" comments are skipped. Keys must decode to exactly 32 bytes. |
| 317 | func ParseMasters(r io.Reader) (*Masters, error) { |
| 318 | m := &Masters{byKid: map[string]Master{}} |
| 319 | sc := bufio.NewScanner(r) |
| 320 | line := 0 |
| 321 | for sc.Scan() { |
| 322 | line++ |
| 323 | s := strings.TrimSpace(sc.Text()) |
| 324 | if s == "" || strings.HasPrefix(s, "#") { |
| 325 | continue |
| 326 | } |
| 327 | f := strings.Fields(s) |
| 328 | if len(f) != 3 { |
| 329 | return nil, fmt.Errorf("relay masters line %d: want 3 fields, got %d", line, len(f)) |
| 330 | } |
| 331 | if !kidRe.MatchString(f[0]) { |
| 332 | return nil, fmt.Errorf("relay masters line %d: bad kid %q", line, f[0]) |
| 333 | } |
| 334 | key, err := base64.StdEncoding.DecodeString(f[1]) |
| 335 | if err != nil || len(key) != 32 { |
| 336 | return nil, fmt.Errorf("relay masters line %d: key must be base64 of 32 bytes", line) |
| 337 | } |
| 338 | var retire time.Time |
| 339 | if f[2] != "-" { |
| 340 | retire, err = time.Parse(time.RFC3339, f[2]) |
| 341 | if err != nil { |
| 342 | return nil, fmt.Errorf("relay masters line %d: bad retire_at %q", line, f[2]) |
| 343 | } |
| 344 | } |
| 345 | m.byKid[f[0]] = Master{Kid: f[0], Key: key, RetireAt: retire} |
| 346 | } |
| 347 | return m, sc.Err() |
| 348 | } |
| 349 | |
| 350 | func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) { |
| 351 | mk, ok := m.byKid[kid] |
| 352 | if !ok || (!mk.RetireAt.IsZero() && !now.Before(mk.RetireAt)) { |
| 353 | return Master{}, false |
| 354 | } |
| 355 | return mk, true |
| 356 | } |
| 357 | |
| 358 | func (m *Masters) Kids() []string { |
| 359 | out := make([]string, 0, len(m.byKid)) |
| 360 | for k := range m.byKid { |
| 361 | out = append(out, k) |
| 362 | } |
| 363 | sort.Strings(out) |
| 364 | return out |
| 365 | } |
| 366 | |
| 367 | // WatchMasters re-reads path whenever its mtime changes, polling every |
| 368 | // five seconds (fsnotify would be a dependency for one file). onChange is |
| 369 | // called with the parse result or error each time the content changes, |
| 370 | // and once at start. |
| 371 | func WatchMasters(ctx context.Context, path string, onChange func(*Masters, error)) { |
| 372 | var last time.Time |
| 373 | tick := time.NewTicker(5 * time.Second) |
| 374 | defer tick.Stop() |
| 375 | load := func() { |
| 376 | st, err := os.Stat(path) |
| 377 | if err != nil { |
| 378 | onChange(nil, err) |
| 379 | return |
| 380 | } |
| 381 | if st.ModTime().Equal(last) { |
| 382 | return |
| 383 | } |
| 384 | last = st.ModTime() |
| 385 | f, err := os.Open(path) |
| 386 | if err != nil { |
| 387 | onChange(nil, err) |
| 388 | return |
| 389 | } |
| 390 | defer f.Close() |
| 391 | onChange(ParseMasters(f)) |
| 392 | } |
| 393 | load() |
| 394 | for { |
| 395 | select { |
| 396 | case <-ctx.Done(): |
| 397 | return |
| 398 | case <-tick.C: |
| 399 | load() |
| 400 | } |
| 401 | } |
| 402 | } |
| 403 | ``` |
| 404 | |
| 405 | - [ ] **Step 4: Run tests, then commit** |
| 406 | |
| 407 | Run: `go test ./internal/relay/ -v` |
| 408 | Expected: PASS. |
| 409 | |
| 410 | ```bash |
| 411 | git add internal/relay/masters.go internal/relay/masters_test.go |
| 412 | git commit -m "relay: masters file with retire_at and hot reload (spec §4)" |
| 413 | ``` |
| 414 | |
| 415 | --- |
| 416 | |
| 417 | ### Task 3: Authority record, lease clock, heartbeat |
| 418 | |
| 419 | **Files:** |
| 420 | - Create: `internal/relay/authority.go` |
| 421 | - Test: `internal/relay/authority_test.go` |
| 422 | |
| 423 | **Interfaces:** |
| 424 | - Produces: |
| 425 | ```go |
| 426 | const AuthorityKey = "control/relay.json" |
| 427 | const LeaseMaxAge = 15 * time.Minute |
| 428 | type AuthorityApp struct{ Account, App, Incarnation string; Regions []string } |
| 429 | type Authority struct{ Apps []AuthorityApp; WrittenAt time.Time } |
| 430 | func (a *Authority) Lookup(account, app string) (AuthorityApp, bool) |
| 431 | func ReadAuthority(ctx, s release.Store) (*Authority, string /*etag*/, error) // os.ErrNotExist → nil,"",nil |
| 432 | type Lease struct{ /* unexported */ } |
| 433 | func (l *Lease) Renew(now time.Time) // on a successful read |
| 434 | func (l *Lease) Live(now time.Time) bool // now - last <= LeaseMaxAge |
| 435 | func HeartbeatKey(box string) string // "control/relay-ack/<box>.json" |
| 436 | type Heartbeat struct{ Box, Region string; WrittenAt time.Time /* of the record held */; ReadAt time.Time; Listeners []string; Kids []string; MastersError string } |
| 437 | func WriteHeartbeat(ctx, s release.Store, hb Heartbeat) error // unconditional Put |
| 438 | ``` |
| 439 | |
| 440 | - [ ] **Step 1: Tests** |
| 441 | |
| 442 | ```go |
| 443 | func TestAuthorityRoundTrip(t *testing.T) { |
| 444 | s := release.NewMemStore() // see internal/release for the in-memory Store used by console tests |
| 445 | ctx := context.Background() |
| 446 | a, _, err := ReadAuthority(ctx, s) |
| 447 | if err != nil || a != nil { |
| 448 | t.Fatalf("absent = %v %v, want nil,nil", a, err) |
| 449 | } |
| 450 | body, _ := json.Marshal(Authority{Apps: []AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}}, WrittenAt: time.Unix(1_800_000_000, 0).UTC()}) |
| 451 | if _, err := s.Put(ctx, AuthorityKey, body, ""); err != nil { |
| 452 | t.Fatal(err) |
| 453 | } |
| 454 | a, etag, err := ReadAuthority(ctx, s) |
| 455 | if err != nil || etag == "" { |
| 456 | t.Fatal(err) |
| 457 | } |
| 458 | if got, ok := a.Lookup("abc", "meet"); !ok || got.Incarnation != "0123456789abcdef" { |
| 459 | t.Fatalf("lookup = %+v %v", got, ok) |
| 460 | } |
| 461 | if _, ok := a.Lookup("abc", "other"); ok { |
| 462 | t.Fatal("unknown app found") |
| 463 | } |
| 464 | } |
| 465 | |
| 466 | func TestLease(t *testing.T) { |
| 467 | var l Lease |
| 468 | t0 := time.Unix(1_800_000_000, 0) |
| 469 | if l.Live(t0) { |
| 470 | t.Fatal("never renewed should not be live") |
| 471 | } |
| 472 | l.Renew(t0) |
| 473 | if !l.Live(t0.Add(14 * time.Minute)) { |
| 474 | t.Fatal("14m should be live") |
| 475 | } |
| 476 | if l.Live(t0.Add(16 * time.Minute)) { |
| 477 | t.Fatal("16m should not be live") |
| 478 | } |
| 479 | } |
| 480 | ``` |
| 481 | |
| 482 | If `release.NewMemStore` does not exist under that name, use whatever in-memory `release.Store` the console tests use (`grep -rn "func New.*Store" internal/release/*.go`); the console fixture's `rel` field is one. |
| 483 | |
| 484 | - [ ] **Step 2: Run to verify failure, then implement** |
| 485 | |
| 486 | ```go |
| 487 | package relay |
| 488 | |
| 489 | import ( |
| 490 | "context" |
| 491 | "encoding/json" |
| 492 | "errors" |
| 493 | "os" |
| 494 | "time" |
| 495 | |
| 496 | "github.com/carlosframework/platform/internal/release" |
| 497 | ) |
| 498 | |
| 499 | const ( |
| 500 | AuthorityKey = "control/relay.json" |
| 501 | LeaseMaxAge = 15 * time.Minute |
| 502 | ) |
| 503 | |
| 504 | type AuthorityApp struct { |
| 505 | Account string `json:"account"` |
| 506 | App string `json:"app"` |
| 507 | Incarnation string `json:"incarnation"` |
| 508 | Regions []string `json:"regions"` |
| 509 | } |
| 510 | |
| 511 | type Authority struct { |
| 512 | Apps []AuthorityApp `json:"apps"` |
| 513 | WrittenAt time.Time `json:"written_at"` |
| 514 | } |
| 515 | |
| 516 | func (a *Authority) Lookup(account, app string) (AuthorityApp, bool) { |
| 517 | if a == nil { |
| 518 | return AuthorityApp{}, false |
| 519 | } |
| 520 | for _, x := range a.Apps { |
| 521 | if x.Account == account && x.App == app { |
| 522 | return x, true |
| 523 | } |
| 524 | } |
| 525 | return AuthorityApp{}, false |
| 526 | } |
| 527 | |
| 528 | func ReadAuthority(ctx context.Context, s release.Store) (*Authority, string, error) { |
| 529 | body, etag, err := s.Get(ctx, AuthorityKey) |
| 530 | if errors.Is(err, os.ErrNotExist) { |
| 531 | return nil, "", nil |
| 532 | } |
| 533 | if err != nil { |
| 534 | return nil, "", err |
| 535 | } |
| 536 | var a Authority |
| 537 | if err := json.Unmarshal(body, &a); err != nil { |
| 538 | return nil, etag, err |
| 539 | } |
| 540 | return &a, etag, nil |
| 541 | } |
| 542 | |
| 543 | // Lease is the relay's serving lease: live while the last SUCCESSFUL |
| 544 | // authority read is younger than LeaseMaxAge. A failed read renews nothing. |
| 545 | type Lease struct{ last time.Time } |
| 546 | |
| 547 | func (l *Lease) Renew(now time.Time) { l.last = now } |
| 548 | func (l *Lease) Live(now time.Time) bool { return !l.last.IsZero() && now.Sub(l.last) <= LeaseMaxAge } |
| 549 | |
| 550 | func HeartbeatKey(box string) string { return "control/relay-ack/" + box + ".json" } |
| 551 | |
| 552 | type Heartbeat struct { |
| 553 | Box string `json:"box"` |
| 554 | Region string `json:"region"` |
| 555 | WrittenAt time.Time `json:"written_at"` // of the authority record held; zero if none |
| 556 | ReadAt time.Time `json:"read_at"` |
| 557 | Listeners []string `json:"listeners"` |
| 558 | Kids []string `json:"kids"` |
| 559 | MastersError string `json:"masters_error,omitempty"` |
| 560 | } |
| 561 | |
| 562 | func WriteHeartbeat(ctx context.Context, s release.Store, hb Heartbeat) error { |
| 563 | body, err := json.Marshal(hb) |
| 564 | if err != nil { |
| 565 | return err |
| 566 | } |
| 567 | _, err = s.Put(ctx, HeartbeatKey(hb.Box), body, "") |
| 568 | return err |
| 569 | } |
| 570 | ``` |
| 571 | |
| 572 | - [ ] **Step 3: Run tests, commit** |
| 573 | |
| 574 | ```bash |
| 575 | git add internal/relay/authority.go internal/relay/authority_test.go |
| 576 | git commit -m "relay: authority record, serving lease, heartbeat (spec §3, §5)" |
| 577 | ``` |
| 578 | |
| 579 | --- |
| 580 | |
| 581 | ### Task 4: Limits |
| 582 | |
| 583 | **Files:** |
| 584 | - Create: `internal/relay/limits.go` |
| 585 | - Test: `internal/relay/limits_test.go` |
| 586 | |
| 587 | **Interfaces:** |
| 588 | - Produces: |
| 589 | ```go |
| 590 | type Limits struct { |
| 591 | AppLive, AppPerMinute int; AppBytesPerHour int64 |
| 592 | AccountLive, AccountPerMinute int; AccountBytesPerHour int64 |
| 593 | BoxLive, BoxPerMinute int |
| 594 | AllocBitsPerSec, BoxBitsPerSec int64 |
| 595 | } |
| 596 | func DefaultLimits() Limits // 64,30,20<<30; 256,60,60<<30; 2048,500; 8_000_000, 200_000_000 |
| 597 | func LimitsFromEnv(getenv func(string) string) Limits // CARLOS_RELAY_LIMIT_* overrides, names below |
| 598 | type Identity struct{ Account, App string } |
| 599 | type Admission struct{ /* unexported; one mutex */ } |
| 600 | func NewAdmission(l Limits, now func() time.Time) *Admission |
| 601 | func (a *Admission) Admit(id Identity) (ok bool, reason string) // counts an attempt even on refusal |
| 602 | func (a *Admission) Release(id Identity) |
| 603 | func (a *Admission) AddBytes(id Identity, n int64) (over bool) // rolling hour; over=true when a budget is crossed |
| 604 | func (a *Admission) Seed(id Identity, at time.Time, n int64) // rebuild rolling window on start |
| 605 | func (a *Admission) OverBudget(id Identity, now time.Time) bool |
| 606 | type Bucket struct{ /* token bucket */ } |
| 607 | func NewBucket(bitsPerSec int64, now func() time.Time) *Bucket |
| 608 | func (b *Bucket) Allow(nbytes int) bool |
| 609 | ``` |
| 610 | - Env names: `CARLOS_RELAY_LIMIT_APP_LIVE`, `_APP_PER_MINUTE`, `_APP_BYTES_PER_HOUR`, `_ACCOUNT_LIVE`, `_ACCOUNT_PER_MINUTE`, `_ACCOUNT_BYTES_PER_HOUR`, `_BOX_LIVE`, `_BOX_PER_MINUTE`, `_ALLOC_BPS`, `_BOX_BPS`. |
| 611 | |
| 612 | - [ ] **Step 1: Tests** |
| 613 | |
| 614 | ```go |
| 615 | func TestAdmissionCaps(t *testing.T) { |
| 616 | now := time.Unix(1_800_000_000, 0) |
| 617 | clock := func() time.Time { return now } |
| 618 | l := DefaultLimits() |
| 619 | l.AppLive, l.AppPerMinute, l.AccountLive, l.BoxLive = 2, 100, 3, 100 |
| 620 | a := NewAdmission(l, clock) |
| 621 | meet := Identity{"abc", "meet"} |
| 622 | docs := Identity{"abc", "docs"} |
| 623 | for i := 0; i < 2; i++ { |
| 624 | if ok, r := a.Admit(meet); !ok { |
| 625 | t.Fatalf("admit %d refused: %s", i, r) |
| 626 | } |
| 627 | } |
| 628 | if ok, r := a.Admit(meet); ok || r != "app live" { |
| 629 | t.Fatalf("third meet = %v %q", ok, r) |
| 630 | } |
| 631 | if ok, _ := a.Admit(docs); !ok { |
| 632 | t.Fatal("docs first should pass") |
| 633 | } |
| 634 | if ok, r := a.Admit(docs); ok || r != "account live" { |
| 635 | t.Fatalf("account cap = %v %q", ok, r) |
| 636 | } |
| 637 | a.Release(meet) |
| 638 | if ok, _ := a.Admit(docs); !ok { |
| 639 | t.Fatal("after release, account has room") |
| 640 | } |
| 641 | } |
| 642 | |
| 643 | func TestAdmissionRate(t *testing.T) { |
| 644 | now := time.Unix(1_800_000_000, 0) |
| 645 | clock := func() time.Time { return now } |
| 646 | l := DefaultLimits() |
| 647 | l.AppPerMinute = 2 |
| 648 | a := NewAdmission(l, clock) |
| 649 | id := Identity{"abc", "meet"} |
| 650 | a.Admit(id); a.Release(id) |
| 651 | a.Admit(id); a.Release(id) |
| 652 | if ok, r := a.Admit(id); ok || r != "app rate" { |
| 653 | t.Fatalf("rate = %v %q", ok, r) |
| 654 | } |
| 655 | now = now.Add(61 * time.Second) |
| 656 | if ok, _ := a.Admit(id); !ok { |
| 657 | t.Fatal("rate window should have rolled") |
| 658 | } |
| 659 | } |
| 660 | |
| 661 | func TestAdmissionBudgetRollsAndSeeds(t *testing.T) { |
| 662 | now := time.Unix(1_800_000_000, 0) |
| 663 | clock := func() time.Time { return now } |
| 664 | l := DefaultLimits() |
| 665 | l.AppBytesPerHour = 100 |
| 666 | a := NewAdmission(l, clock) |
| 667 | id := Identity{"abc", "meet"} |
| 668 | if a.AddBytes(id, 60) { |
| 669 | t.Fatal("60 under 100") |
| 670 | } |
| 671 | if !a.AddBytes(id, 50) { |
| 672 | t.Fatal("110 over 100") |
| 673 | } |
| 674 | now = now.Add(61 * time.Minute) |
| 675 | if a.OverBudget(id, now) { |
| 676 | t.Fatal("window rolled") |
| 677 | } |
| 678 | b := NewAdmission(l, clock) |
| 679 | b.Seed(id, now.Add(-30*time.Minute), 90) |
| 680 | if !b.AddBytes(id, 20) { |
| 681 | t.Fatal("seeded 90 + 20 over 100") |
| 682 | } |
| 683 | } |
| 684 | |
| 685 | func TestBucket(t *testing.T) { |
| 686 | now := time.Unix(1_800_000_000, 0) |
| 687 | clock := func() time.Time { return now } |
| 688 | b := NewBucket(8000, clock) // 1000 bytes/s |
| 689 | if !b.Allow(1000) { |
| 690 | t.Fatal("first second's worth") |
| 691 | } |
| 692 | if b.Allow(1) { |
| 693 | t.Fatal("empty") |
| 694 | } |
| 695 | now = now.Add(500 * time.Millisecond) |
| 696 | if !b.Allow(500) || b.Allow(1) { |
| 697 | t.Fatal("refilled 500") |
| 698 | } |
| 699 | } |
| 700 | ``` |
| 701 | |
| 702 | - [ ] **Step 2: Implement** |
| 703 | |
| 704 | ```go |
| 705 | package relay |
| 706 | |
| 707 | import ( |
| 708 | "strconv" |
| 709 | "sync" |
| 710 | "time" |
| 711 | ) |
| 712 | |
| 713 | type Limits struct { |
| 714 | AppLive, AppPerMinute int |
| 715 | AppBytesPerHour int64 |
| 716 | AccountLive, AccountPerMinute int |
| 717 | AccountBytesPerHour int64 |
| 718 | BoxLive, BoxPerMinute int |
| 719 | AllocBitsPerSec, BoxBitsPerSec int64 |
| 720 | } |
| 721 | |
| 722 | func DefaultLimits() Limits { |
| 723 | return Limits{ |
| 724 | AppLive: 64, AppPerMinute: 30, AppBytesPerHour: 20 << 30, |
| 725 | AccountLive: 256, AccountPerMinute: 60, AccountBytesPerHour: 60 << 30, |
| 726 | BoxLive: 2048, BoxPerMinute: 500, |
| 727 | AllocBitsPerSec: 8_000_000, BoxBitsPerSec: 200_000_000, |
| 728 | } |
| 729 | } |
| 730 | |
| 731 | func LimitsFromEnv(getenv func(string) string) Limits { |
| 732 | l := DefaultLimits() |
| 733 | i := func(name string, dst *int) { |
| 734 | if v, err := strconv.Atoi(getenv("CARLOS_RELAY_LIMIT_" + name)); err == nil && v > 0 { |
| 735 | *dst = v |
| 736 | } |
| 737 | } |
| 738 | i64 := func(name string, dst *int64) { |
| 739 | if v, err := strconv.ParseInt(getenv("CARLOS_RELAY_LIMIT_"+name), 10, 64); err == nil && v > 0 { |
| 740 | *dst = v |
| 741 | } |
| 742 | } |
| 743 | i("APP_LIVE", &l.AppLive); i("APP_PER_MINUTE", &l.AppPerMinute); i64("APP_BYTES_PER_HOUR", &l.AppBytesPerHour) |
| 744 | i("ACCOUNT_LIVE", &l.AccountLive); i("ACCOUNT_PER_MINUTE", &l.AccountPerMinute); i64("ACCOUNT_BYTES_PER_HOUR", &l.AccountBytesPerHour) |
| 745 | i("BOX_LIVE", &l.BoxLive); i("BOX_PER_MINUTE", &l.BoxPerMinute) |
| 746 | i64("ALLOC_BPS", &l.AllocBitsPerSec); i64("BOX_BPS", &l.BoxBitsPerSec) |
| 747 | return l |
| 748 | } |
| 749 | |
| 750 | type Identity struct{ Account, App string } |
| 751 | |
| 752 | type window struct { // sliding counters bucketed per minute, 60 buckets for an hour |
| 753 | at [60]time.Time |
| 754 | n [60]int64 |
| 755 | } |
| 756 | |
| 757 | func (w *window) add(now time.Time, n int64) { |
| 758 | m := now.Truncate(time.Minute) |
| 759 | i := int(m.Unix()/60) % 60 |
| 760 | if !w.at[i].Equal(m) { |
| 761 | w.at[i], w.n[i] = m, 0 |
| 762 | } |
| 763 | w.n[i] += n |
| 764 | } |
| 765 | |
| 766 | func (w *window) sum(now time.Time, span time.Duration) int64 { |
| 767 | var s int64 |
| 768 | cut := now.Add(-span) |
| 769 | for i := range w.at { |
| 770 | if !w.at[i].IsZero() && w.at[i].After(cut) { |
| 771 | s += w.n[i] |
| 772 | } |
| 773 | } |
| 774 | return s |
| 775 | } |
| 776 | |
| 777 | type counters struct { |
| 778 | live int |
| 779 | starts window // attempts per minute |
| 780 | bytes window // bytes per rolling hour |
| 781 | } |
| 782 | |
| 783 | type Admission struct { |
| 784 | mu sync.Mutex |
| 785 | l Limits |
| 786 | now func() time.Time |
| 787 | app map[Identity]*counters |
| 788 | acct map[string]*counters |
| 789 | box counters |
| 790 | } |
| 791 | |
| 792 | func NewAdmission(l Limits, now func() time.Time) *Admission { |
| 793 | return &Admission{l: l, now: now, app: map[Identity]*counters{}, acct: map[string]*counters{}} |
| 794 | } |
| 795 | |
| 796 | func (a *Admission) get(id Identity) (*counters, *counters) { |
| 797 | c, ok := a.app[id] |
| 798 | if !ok { |
| 799 | c = &counters{} |
| 800 | a.app[id] = c |
| 801 | } |
| 802 | ac, ok := a.acct[id.Account] |
| 803 | if !ok { |
| 804 | ac = &counters{} |
| 805 | a.acct[id.Account] = ac |
| 806 | } |
| 807 | return c, ac |
| 808 | } |
| 809 | |
| 810 | // Admit counts an attempt for the app, the account and the box whether or |
| 811 | // not it is admitted, then checks live caps, rates and budgets in that order. |
| 812 | func (a *Admission) Admit(id Identity) (bool, string) { |
| 813 | a.mu.Lock() |
| 814 | defer a.mu.Unlock() |
| 815 | now := a.now() |
| 816 | c, ac := a.get(id) |
| 817 | c.starts.add(now, 1); ac.starts.add(now, 1); a.box.starts.add(now, 1) |
| 818 | switch { |
| 819 | case c.live >= a.l.AppLive: |
| 820 | return false, "app live" |
| 821 | case ac.live >= a.l.AccountLive: |
| 822 | return false, "account live" |
| 823 | case a.box.live >= a.l.BoxLive: |
| 824 | return false, "box live" |
| 825 | case c.starts.sum(now, time.Minute) > int64(a.l.AppPerMinute): |
| 826 | return false, "app rate" |
| 827 | case ac.starts.sum(now, time.Minute) > int64(a.l.AccountPerMinute): |
| 828 | return false, "account rate" |
| 829 | case a.box.starts.sum(now, time.Minute) > int64(a.l.BoxPerMinute): |
| 830 | return false, "box rate" |
| 831 | case c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour: |
| 832 | return false, "app budget" |
| 833 | case ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour: |
| 834 | return false, "account budget" |
| 835 | } |
| 836 | c.live++; ac.live++; a.box.live++ |
| 837 | return true, "" |
| 838 | } |
| 839 | |
| 840 | func (a *Admission) Release(id Identity) { |
| 841 | a.mu.Lock() |
| 842 | defer a.mu.Unlock() |
| 843 | c, ac := a.get(id) |
| 844 | if c.live > 0 { c.live-- } |
| 845 | if ac.live > 0 { ac.live-- } |
| 846 | if a.box.live > 0 { a.box.live-- } |
| 847 | } |
| 848 | |
| 849 | // AddBytes records forwarded bytes and reports whether the app or account |
| 850 | // rolling-hour budget is now exceeded. |
| 851 | func (a *Admission) AddBytes(id Identity, n int64) bool { |
| 852 | a.mu.Lock() |
| 853 | defer a.mu.Unlock() |
| 854 | now := a.now() |
| 855 | c, ac := a.get(id) |
| 856 | c.bytes.add(now, n); ac.bytes.add(now, n) |
| 857 | return c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour || ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour |
| 858 | } |
| 859 | |
| 860 | func (a *Admission) Seed(id Identity, at time.Time, n int64) { |
| 861 | a.mu.Lock() |
| 862 | defer a.mu.Unlock() |
| 863 | c, ac := a.get(id) |
| 864 | c.bytes.add(at, n); ac.bytes.add(at, n) |
| 865 | } |
| 866 | |
| 867 | func (a *Admission) OverBudget(id Identity, now time.Time) bool { |
| 868 | a.mu.Lock() |
| 869 | defer a.mu.Unlock() |
| 870 | c, ac := a.get(id) |
| 871 | return c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour || ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour |
| 872 | } |
| 873 | |
| 874 | // Bucket is a token bucket in bytes with a one-second burst. |
| 875 | type Bucket struct { |
| 876 | mu sync.Mutex |
| 877 | rate float64 // bytes per second |
| 878 | tokens float64 |
| 879 | last time.Time |
| 880 | now func() time.Time |
| 881 | } |
| 882 | |
| 883 | func NewBucket(bitsPerSec int64, now func() time.Time) *Bucket { |
| 884 | r := float64(bitsPerSec) / 8 |
| 885 | return &Bucket{rate: r, tokens: r, last: now(), now: now} |
| 886 | } |
| 887 | |
| 888 | func (b *Bucket) Allow(nbytes int) bool { |
| 889 | b.mu.Lock() |
| 890 | defer b.mu.Unlock() |
| 891 | t := b.now() |
| 892 | b.tokens += t.Sub(b.last).Seconds() * b.rate |
| 893 | if b.tokens > b.rate { |
| 894 | b.tokens = b.rate |
| 895 | } |
| 896 | b.last = t |
| 897 | if float64(nbytes) > b.tokens { |
| 898 | return false |
| 899 | } |
| 900 | b.tokens -= float64(nbytes) |
| 901 | return true |
| 902 | } |
| 903 | ``` |
| 904 | |
| 905 | - [ ] **Step 3: Run tests, commit** |
| 906 | |
| 907 | ```bash |
| 908 | git add internal/relay/limits.go internal/relay/limits_test.go |
| 909 | git commit -m "relay: admission, rolling budgets, token buckets (spec §3 limits)" |
| 910 | ``` |
| 911 | |
| 912 | --- |
| 913 | |
| 914 | ### Task 5: Wrapped sockets: counting, buckets, EIP port gate, close |
| 915 | |
| 916 | **Files:** |
| 917 | - Create: `internal/relay/conn.go` |
| 918 | - Test: `internal/relay/conn_test.go` |
| 919 | |
| 920 | **Interfaces:** |
| 921 | - Produces: |
| 922 | ```go |
| 923 | // Binding is what a relay socket belongs to. |
| 924 | type Binding struct{ ID Identity; Expiry time.Time; Kid string; Src string /* typed source key */ } |
| 925 | // relayConn wraps an allocation's relay PacketConn. |
| 926 | type relayConn struct{ /* unexported */ } |
| 927 | func newRelayConn(inner net.PacketConn, b *Binding, eip net.IP, ports *portSet, appBucket, boxBucket *Bucket, onIn func(Identity, int), onRejected func(Identity, int), onClose func(*relayConn)) *relayConn |
| 928 | // WriteTo: client→peer forwarding. Drops (returns n,nil) when the bucket refuses or when dst is the EIP on a port not in ports. |
| 929 | // ReadFrom: peer→relay; counted by the server only after pion forwards (see Task 6), so this wrapper counts nothing on read. |
| 930 | // Close: idempotent; runs onClose once. |
| 931 | type portSet struct{ /* unexported; mutex + map[int]bool */ } |
| 932 | func (p *portSet) add(port int); func (p *portSet) remove(port int); func (p *portSet) has(port int) bool |
| 933 | func typedSrc(a net.Addr) string // "udp:" or "tcp:" + a.String(); the owner-binding key |
| 934 | ``` |
| 935 | |
| 936 | - [ ] **Step 1: Tests** |
| 937 | |
| 938 | ```go |
| 939 | func TestRelayConnGatesEIPPorts(t *testing.T) { |
| 940 | inner := &fakePacketConn{} |
| 941 | ports := &portSet{m: map[int]bool{50000: true}} |
| 942 | eip := net.ParseIP("203.0.113.5") |
| 943 | now := time.Unix(1_800_000_000, 0) |
| 944 | clock := func() time.Time { return now } |
| 945 | var in int |
| 946 | c := newRelayConn(inner, &Binding{ID: Identity{"abc", "meet"}}, eip, ports, NewBucket(1<<40, clock), NewBucket(1<<40, clock), |
| 947 | func(_ Identity, n int) { in += n }, nil, func(*relayConn) {}) |
| 948 | ok := &net.UDPAddr{IP: eip, Port: 50000} |
| 949 | bad := &net.UDPAddr{IP: eip, Port: 3478} |
| 950 | other := &net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 4444} |
| 951 | c.WriteTo([]byte("abc"), ok) |
| 952 | c.WriteTo([]byte("abc"), bad) |
| 953 | c.WriteTo([]byte("abcd"), other) |
| 954 | if len(inner.written) != 2 || in != 7 { |
| 955 | t.Fatalf("written=%d in=%d, want 2 packets and 7 bytes", len(inner.written), in) |
| 956 | } |
| 957 | } |
| 958 | |
| 959 | func TestRelayConnBucketDrops(t *testing.T) { |
| 960 | inner := &fakePacketConn{} |
| 961 | now := time.Unix(1_800_000_000, 0) |
| 962 | clock := func() time.Time { return now } |
| 963 | c := newRelayConn(inner, &Binding{}, nil, &portSet{m: map[int]bool{}}, NewBucket(80, clock), NewBucket(1<<40, clock), func(Identity, int) {}, nil, func(*relayConn) {}) |
| 964 | dst := &net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 4444} |
| 965 | c.WriteTo(make([]byte, 10), dst) // 10 bytes = the whole second |
| 966 | c.WriteTo(make([]byte, 1), dst) |
| 967 | if len(inner.written) != 1 { |
| 968 | t.Fatalf("written = %d, want 1 (second dropped by bucket)", len(inner.written)) |
| 969 | } |
| 970 | } |
| 971 | |
| 972 | func TestRelayConnCloseOnce(t *testing.T) { |
| 973 | n := 0 |
| 974 | c := newRelayConn(&fakePacketConn{}, &Binding{}, nil, &portSet{m: map[int]bool{}}, nil, nil, nil, nil, func(*relayConn) { n++ }) |
| 975 | c.Close(); c.Close() |
| 976 | if n != 1 { |
| 977 | t.Fatalf("onClose ran %d times", n) |
| 978 | } |
| 979 | } |
| 980 | |
| 981 | func TestTypedSrc(t *testing.T) { |
| 982 | u := &net.UDPAddr{IP: net.ParseIP("192.0.2.1"), Port: 5000} |
| 983 | tc := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 5000} |
| 984 | if typedSrc(u) == typedSrc(tc) { |
| 985 | t.Fatal("udp and tcp with the same address must not collide") |
| 986 | } |
| 987 | } |
| 988 | |
| 989 | // fakePacketConn records writes; ReadFrom blocks until Close. |
| 990 | type fakePacketConn struct { |
| 991 | mu sync.Mutex |
| 992 | written [][]byte |
| 993 | closed chan struct{} |
| 994 | once sync.Once |
| 995 | } |
| 996 | func (f *fakePacketConn) init() { f.once.Do(func() { f.closed = make(chan struct{}) }) } |
| 997 | func (f *fakePacketConn) ReadFrom(b []byte) (int, net.Addr, error) { f.init(); <-f.closed; return 0, nil, net.ErrClosed } |
| 998 | func (f *fakePacketConn) WriteTo(b []byte, a net.Addr) (int, error) { f.mu.Lock(); defer f.mu.Unlock(); f.written = append(f.written, append([]byte(nil), b...)); return len(b), nil } |
| 999 | func (f *fakePacketConn) Close() error { f.init(); select { case <-f.closed: default: close(f.closed) }; return nil } |
| 1000 | func (f *fakePacketConn) LocalAddr() net.Addr { return &net.UDPAddr{IP: net.IPv4zero, Port: 50000} } |
| 1001 | func (f *fakePacketConn) SetDeadline(time.Time) error { return nil } |
| 1002 | func (f *fakePacketConn) SetReadDeadline(time.Time) error { return nil } |
| 1003 | func (f *fakePacketConn) SetWriteDeadline(time.Time) error { return nil } |
| 1004 | ``` |
| 1005 | |
| 1006 | - [ ] **Step 2: Implement** |
| 1007 | |
| 1008 | ```go |
| 1009 | package relay |
| 1010 | |
| 1011 | import ( |
| 1012 | "net" |
| 1013 | "sync" |
| 1014 | "time" |
| 1015 | ) |
| 1016 | |
| 1017 | // Binding is the identity an allocation was created under. Src is the |
| 1018 | // typed source key (typedSrc) pion will present on later requests. |
| 1019 | type Binding struct { |
| 1020 | ID Identity |
| 1021 | Expiry time.Time |
| 1022 | Kid string |
| 1023 | Src string |
| 1024 | } |
| 1025 | |
| 1026 | // typedSrc keys the owner binding on transport AND address, because pion |
| 1027 | // tags every allocation's tuple as UDP even for TCP clients and a UDP and a |
| 1028 | // TCP client can share an address string. |
| 1029 | func typedSrc(a net.Addr) string { |
| 1030 | switch a.(type) { |
| 1031 | case *net.TCPAddr: |
| 1032 | return "tcp:" + a.String() |
| 1033 | default: |
| 1034 | return "udp:" + a.String() |
| 1035 | } |
| 1036 | } |
| 1037 | |
| 1038 | type portSet struct { |
| 1039 | mu sync.Mutex |
| 1040 | m map[int]bool |
| 1041 | } |
| 1042 | |
| 1043 | func (p *portSet) add(port int) { p.mu.Lock(); p.m[port] = true; p.mu.Unlock() } |
| 1044 | func (p *portSet) remove(port int) { p.mu.Lock(); delete(p.m, port); p.mu.Unlock() } |
| 1045 | func (p *portSet) has(port int) bool { p.mu.Lock(); defer p.mu.Unlock(); return p.m[port] } |
| 1046 | |
| 1047 | // relayConn is the PacketConn the relay hands pion for one allocation. |
| 1048 | // Closing it is how the relay ends the allocation: pion's read loop deletes |
| 1049 | // the allocation when ReadFrom errors (allocation.go:301 in pion/turn |
| 1050 | // v4.1.4). WriteTo is the client→peer leg and is where forwarded bytes are |
| 1051 | // counted, the per-allocation and box buckets applied, and the EIP port |
| 1052 | // gate enforced. |
| 1053 | type relayConn struct { |
| 1054 | net.PacketConn |
| 1055 | b *Binding |
| 1056 | eip net.IP |
| 1057 | ports *portSet |
| 1058 | appBucket *Bucket |
| 1059 | boxBucket *Bucket |
| 1060 | onIn func(Identity, int) |
| 1061 | onRejected func(Identity, int) |
| 1062 | onClose func(*relayConn) |
| 1063 | once sync.Once |
| 1064 | } |
| 1065 | |
| 1066 | func newRelayConn(inner net.PacketConn, b *Binding, eip net.IP, ports *portSet, appBucket, boxBucket *Bucket, |
| 1067 | onIn func(Identity, int), onRejected func(Identity, int), onClose func(*relayConn)) *relayConn { |
| 1068 | return &relayConn{PacketConn: inner, b: b, eip: eip, ports: ports, appBucket: appBucket, boxBucket: boxBucket, |
| 1069 | onIn: onIn, onRejected: onRejected, onClose: onClose} |
| 1070 | } |
| 1071 | |
| 1072 | func (c *relayConn) WriteTo(p []byte, addr net.Addr) (int, error) { |
| 1073 | if ua, ok := addr.(*net.UDPAddr); ok && c.eip != nil && ua.IP.Equal(c.eip) && !c.ports.has(ua.Port) { |
| 1074 | if c.onRejected != nil { |
| 1075 | c.onRejected(c.b.ID, len(p)) |
| 1076 | } |
| 1077 | return len(p), nil // dropped: only live relay ports are reachable on our own EIP |
| 1078 | } |
| 1079 | if (c.appBucket != nil && !c.appBucket.Allow(len(p))) || (c.boxBucket != nil && !c.boxBucket.Allow(len(p))) { |
| 1080 | return len(p), nil // dropped by a token bucket |
| 1081 | } |
| 1082 | n, err := c.PacketConn.WriteTo(p, addr) |
| 1083 | if n > 0 && c.onIn != nil { |
| 1084 | c.onIn(c.b.ID, n) |
| 1085 | } |
| 1086 | return n, err |
| 1087 | } |
| 1088 | |
| 1089 | func (c *relayConn) Close() error { |
| 1090 | var err error |
| 1091 | c.once.Do(func() { |
| 1092 | err = c.PacketConn.Close() |
| 1093 | if c.onClose != nil { |
| 1094 | c.onClose(c) |
| 1095 | } |
| 1096 | }) |
| 1097 | return err |
| 1098 | } |
| 1099 | ``` |
| 1100 | |
| 1101 | - [ ] **Step 3: Run tests, commit** |
| 1102 | |
| 1103 | ```bash |
| 1104 | git add internal/relay/conn.go internal/relay/conn_test.go |
| 1105 | git commit -m "relay: wrapped relay socket with counting, buckets and the EIP port gate (spec §3)" |
| 1106 | ``` |
| 1107 | |
| 1108 | --- |
| 1109 | |
| 1110 | ### Task 6: The pion server wrapper (Opus) |
| 1111 | |
| 1112 | **Files:** |
| 1113 | - Modify: `go.mod`, `go.sum` (add `github.com/pion/turn/v4 v4.1.4`) |
| 1114 | - Create: `internal/relay/server.go` |
| 1115 | - Test: `internal/relay/server_test.go` (end to end on loopback) |
| 1116 | |
| 1117 | **Interfaces:** |
| 1118 | - Consumes: Tasks 1 to 5. |
| 1119 | - Produces: |
| 1120 | ```go |
| 1121 | type ServerConfig struct { |
| 1122 | Region string |
| 1123 | RelayIP net.IP // the EIP advertised to clients |
| 1124 | UDP, TCP string // listen addresses, e.g. ":3478"; empty disables |
| 1125 | MinPort, MaxPort uint16 // 49152, 65535 |
| 1126 | DenyCIDRs []*net.IPNet // from CARLOS_RELAY_DENY_CIDRS plus the built-in private ranges |
| 1127 | Masters func() *Masters |
| 1128 | Authority func() *Authority |
| 1129 | Lease func(now time.Time) bool |
| 1130 | Limits Limits |
| 1131 | Now func() time.Time |
| 1132 | Log func(string, ...any) |
| 1133 | Meter *Meter // Task 7; nil allowed in tests |
| 1134 | } |
| 1135 | type Server struct{ /* unexported */ } |
| 1136 | func NewServer(cfg ServerConfig) (*Server, error) |
| 1137 | func (s *Server) Close() error |
| 1138 | func (s *Server) EndAll(reason string) // lease lapse, shedding |
| 1139 | func (s *Server) EndWhere(func(*Binding) bool, reason string) // disable, incarnation change, region removed, budget, master retired |
| 1140 | func (s *Server) Live() int |
| 1141 | ``` |
| 1142 | |
| 1143 | - [ ] **Step 1: Add the dependency** (sandbox lifted for this one command) |
| 1144 | |
| 1145 | Run: `GOFLAGS= GOPROXY=https://proxy.golang.org go get github.com/pion/turn/v4@v4.1.4 && go mod tidy` |
| 1146 | Then: `GOFLAGS=-mod=mod GOPROXY=off go build ./...` must still pass; `git diff go.mod` shows only pion modules added. |
| 1147 | |
| 1148 | - [ ] **Step 2: Write the end-to-end test** |
| 1149 | |
| 1150 | ```go |
| 1151 | func TestServerEndToEnd(t *testing.T) { |
| 1152 | master := sha256.Sum256([]byte("carlos-relay-test-master")) |
| 1153 | masters, _ := ParseMasters(strings.NewReader("k1 " + b64(master[:]) + " -\n")) |
| 1154 | auth := &Authority{Apps: []AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"test"}}}, WrittenAt: time.Now()} |
| 1155 | now := time.Now() |
| 1156 | var mu sync.Mutex |
| 1157 | clock := func() time.Time { mu.Lock(); defer mu.Unlock(); return now } |
| 1158 | s, err := NewServer(ServerConfig{ |
| 1159 | Region: "test", RelayIP: net.ParseIP("127.0.0.1"), UDP: "127.0.0.1:0", |
| 1160 | MinPort: 40000, MaxPort: 40100, |
| 1161 | Masters: func() *Masters { return masters }, Authority: func() *Authority { return auth }, |
| 1162 | Lease: func(time.Time) bool { return true }, Limits: DefaultLimits(), Now: clock, Log: t.Logf, |
| 1163 | }) |
| 1164 | if err != nil { |
| 1165 | t.Fatal(err) |
| 1166 | } |
| 1167 | defer s.Close() |
| 1168 | |
| 1169 | key := DeriveKey(master[:], "abc", "meet", "0123456789abcdef") |
| 1170 | exp := now.Add(time.Hour).Unix() |
| 1171 | username := fmt.Sprintf("%d:%s", exp, Suffix("abc", "meet", "0123456789abcdef", "k1")) |
| 1172 | client := turnClient(t, s.UDPAddr(), username, Credential(key, username)) |
| 1173 | relayConn, err := client.Allocate() |
| 1174 | if err != nil { |
| 1175 | t.Fatalf("allocate: %v", err) |
| 1176 | } |
| 1177 | defer relayConn.Close() |
| 1178 | if s.Live() != 1 { |
| 1179 | t.Fatalf("live = %d", s.Live()) |
| 1180 | } |
| 1181 | |
| 1182 | // Wrong app's credential from the same source is refused (owner binding). |
| 1183 | other := fmt.Sprintf("%d:%s", exp, Suffix("abc", "docs", "0123456789abcdef", "k1")) |
| 1184 | if _, err := turnClient(t, s.UDPAddr(), other, Credential(DeriveKey(master[:], "abc", "docs", "0123456789abcdef"), other)).Allocate(); err == nil { |
| 1185 | t.Fatal("unknown app allocated") |
| 1186 | } |
| 1187 | |
| 1188 | // Expiry ends the allocation with no client action. |
| 1189 | mu.Lock(); now = now.Add(2 * time.Hour); mu.Unlock() |
| 1190 | s.tick() // exported for tests: runs the expiry sweep once |
| 1191 | waitFor(t, func() bool { return s.Live() == 0 }) |
| 1192 | } |
| 1193 | |
| 1194 | func TestServerRefusesPrivatePeer(t *testing.T) { |
| 1195 | // allocate as above, then client.CreatePermission(&net.UDPAddr{IP: net.ParseIP("10.0.0.1"), Port: 1}) must error. |
| 1196 | } |
| 1197 | |
| 1198 | func TestServerEndWhere(t *testing.T) { |
| 1199 | // allocate, then s.EndWhere(func(b *Binding) bool { return b.ID.App == "meet" }, "disable"); waitFor Live()==0. |
| 1200 | } |
| 1201 | ``` |
| 1202 | |
| 1203 | `turnClient` builds a `turn.Client` with `turn.ClientConfig{STUNServerAddr, TURNServerAddr: addr, Conn: udp, Username, Password, Realm: "carlos"}` and calls `Listen()`; see `github.com/pion/turn/v4` examples `turn-client/udp`. `waitFor` polls for two seconds. |
| 1204 | |
| 1205 | - [ ] **Step 3: Implement `server.go`** |
| 1206 | |
| 1207 | Design, from spec §3: |
| 1208 | |
| 1209 | - `NewServer` opens the UDP and TCP listeners, builds a `turn.ServerConfig{Realm: "carlos", AuthHandler: s.auth, QuotaHandler: s.quota, EventHandler: turn.EventHandler{OnAllocationCreated: s.created, OnAllocationDeleted: s.deleted}, PacketConnConfigs: [{PacketConn: udp, RelayAddressGenerator: s, PermissionHandler: s.permit}], ListenerConfigs: [{Listener: tcp, RelayAddressGenerator: s, PermissionHandler: s.permit}]}`. |
| 1210 | - `Server` implements `turn.RelayAddressGenerator`: `Validate()`, `AllocatePacketConn(network string, requestedPort int) (net.PacketConn, net.Addr, error)` binds a UDP socket on `RelayIP`-side interface in `[MinPort, MaxPort]`, wraps it in `newRelayConn`, records it in `s.pending` keyed by relay port, adds the port to `s.ports`, and returns the wrapper and `&net.UDPAddr{IP: cfg.RelayIP, Port: port}`. `AllocateConn` returns `errUnsupported`. |
| 1211 | - `auth(username, realm string, src net.Addr) ([]byte, bool)`: `ParseUsername`; expiry in `(now, now+MaxCredentialLife]`; `s.bound[typedSrc(src)]` if present must have the same Identity+Incarnation; `Authority().Lookup`; incarnation and region match; `Masters().Lookup(kid, now)`; `DeriveKey`; `Credential`; `turn.GenerateAuthKey(username, realm, cred)`; stash the parsed username in `s.lastAuth[typedSrc(src)]` so `created` can build the Binding. Count refusal reasons per app in `s.refusals`. |
| 1212 | - `quota(username, realm string, src net.Addr) bool`: `Lease(now)` false → refuse; `Admit(Identity)` → refuse with reason logged. |
| 1213 | - `created(src, dst net.Addr, proto, username, realm string, relayAddr net.Addr)`: take `s.pending[relayPort]`, set its Binding from `s.lastAuth[typedSrc(src)]`, move it to `s.allocs[typedSrc(src)]`, set `s.bound[typedSrc(src)] = binding`, start the expiry timer (`time.AfterFunc(expiry-now, conn.Close)`). |
| 1214 | - `deleted(...)`: no-op; cleanup is in `relayConn.onClose`: remove the port from `s.ports`, `Admission.Release`, delete `s.bound[src]` only if it still points at this conn's binding, delete `s.allocs[src]` likewise. |
| 1215 | - `permit(client net.Addr, peer net.IP) bool`: refuse if `peer` is in any `DenyCIDRs`, loopback, link-local, RFC 1918, ULA, or equals a relay listener's private address; allow `RelayIP` (the EIP; the port gate is in `relayConn.WriteTo`). |
| 1216 | - `EndWhere`/`EndAll`: iterate `s.allocs`, `Close()` matching conns. |
| 1217 | - `tick()`: close any conn whose `Binding.Expiry <= now` (belt for the timers), and any whose master is retired (`Masters().Lookup(kid, now)` false), and any whose app is `OverBudget`. |
| 1218 | - Bytes: `onIn` → `Admission.AddBytes`; if over, `EndWhere(app)`. The `out` leg is counted by wrapping the listener conns' `WriteTo` (a `listenerConn` type in this file mirroring `relayConn` without the EIP gate, attributing by `s.bound[typedSrc(dst)]`). Both legs call `Meter.Add(id, in, out)` when `Meter != nil`. |
| 1219 | - Built-in deny ranges: `127.0.0.0/8, ::1/128, 169.254.0.0/16, fe80::/10, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, fc00::/7, 100.64.0.0/10`. |
| 1220 | |
| 1221 | - [ ] **Step 4: Run the end-to-end tests, then the cross-compile check** |
| 1222 | |
| 1223 | Run: `go test ./internal/relay/ -run TestServer -v` |
| 1224 | Run: `for t in darwin/arm64 windows/amd64 linux/arm64; do GOOS=${t%/*} GOARCH=${t#*/} CGO_ENABLED=0 go build ./... || echo FAIL $t; done` |
| 1225 | Expected: PASS, no FAIL lines. |
| 1226 | |
| 1227 | - [ ] **Step 5: Commit** |
| 1228 | |
| 1229 | ```bash |
| 1230 | git add go.mod go.sum internal/relay/server.go internal/relay/server_test.go |
| 1231 | git commit -m "relay: pion server with our own address generator, owner binding, termination and peer policy (spec §3, §4)" |
| 1232 | ``` |
| 1233 | |
| 1234 | --- |
| 1235 | |
| 1236 | ### Task 7: Meter |
| 1237 | |
| 1238 | **Files:** |
| 1239 | - Create: `internal/relay/meter.go` |
| 1240 | - Test: `internal/relay/meter_test.go` |
| 1241 | |
| 1242 | **Interfaces:** |
| 1243 | - Produces: |
| 1244 | ```go |
| 1245 | func MeterKey(box, hour string) string // "metrics/relay/<box>/<hour>.json" |
| 1246 | type MeterSample struct{ At time.Time; Apps map[string]MeterApp } // key "<account>/<app>" |
| 1247 | type MeterApp struct{ In, Out, Rejected int64; Allocs int } |
| 1248 | type MeterChunk struct{ V int; Box, Hour string; Samples []MeterSample } |
| 1249 | type Meter struct{ /* unexported */ } |
| 1250 | func NewMeter(box string, now func() time.Time) *Meter |
| 1251 | func (m *Meter) Add(id Identity, in, out, rejected int64) |
| 1252 | func (m *Meter) SetAllocs(id Identity, n int) |
| 1253 | func (m *Meter) Flush(ctx, s release.Store) error // one sample of deltas since last flush; read-append-CAS on the hour's object, retry 3x |
| 1254 | func LoadWindow(ctx, s release.Store, box string, now time.Time) ([]MeterSample, error) // previous and current hour's samples |
| 1255 | ``` |
| 1256 | |
| 1257 | - [ ] **Step 1: Tests** |
| 1258 | |
| 1259 | ```go |
| 1260 | func TestMeterFlushAppendsDeltas(t *testing.T) { |
| 1261 | s := newMemStore() |
| 1262 | now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC) |
| 1263 | m := NewMeter("box1", func() time.Time { return now }) |
| 1264 | id := Identity{"abc", "meet"} |
| 1265 | m.Add(id, 100, 200, 5) |
| 1266 | if err := m.Flush(context.Background(), s); err != nil { t.Fatal(err) } |
| 1267 | m.Add(id, 10, 20, 0) |
| 1268 | if err := m.Flush(context.Background(), s); err != nil { t.Fatal(err) } |
| 1269 | body, _, _ := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12")) |
| 1270 | var c MeterChunk |
| 1271 | json.Unmarshal(body, &c) |
| 1272 | if len(c.Samples) != 2 || c.Samples[0].Apps["abc/meet"].In != 100 || c.Samples[1].Apps["abc/meet"].Out != 20 { |
| 1273 | t.Fatalf("chunk = %+v", c) |
| 1274 | } |
| 1275 | } |
| 1276 | |
| 1277 | func TestLoadWindowSpansTwoHours(t *testing.T) { |
| 1278 | s := newMemStore() |
| 1279 | now := time.Date(2026, 9, 7, 12, 1, 0, 0, time.UTC) |
| 1280 | m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }) // 11:31 |
| 1281 | m.Add(Identity{"abc", "meet"}, 7, 0, 0) |
| 1282 | m.Flush(context.Background(), s) |
| 1283 | samples, err := LoadWindow(context.Background(), s, "box1", now) |
| 1284 | if err != nil || len(samples) != 1 || samples[0].Apps["abc/meet"].In != 7 { |
| 1285 | t.Fatalf("window = %+v %v", samples, err) |
| 1286 | } |
| 1287 | } |
| 1288 | ``` |
| 1289 | |
| 1290 | - [ ] **Step 2: Implement** |
| 1291 | |
| 1292 | ```go |
| 1293 | package relay |
| 1294 | |
| 1295 | import ( |
| 1296 | "context" |
| 1297 | "encoding/json" |
| 1298 | "errors" |
| 1299 | "os" |
| 1300 | "sync" |
| 1301 | "time" |
| 1302 | |
| 1303 | "github.com/carlosframework/platform/internal/metrics" |
| 1304 | "github.com/carlosframework/platform/internal/release" |
| 1305 | ) |
| 1306 | |
| 1307 | func MeterKey(box, hour string) string { return "metrics/relay/" + box + "/" + hour + ".json" } |
| 1308 | |
| 1309 | type MeterApp struct { |
| 1310 | In int64 `json:"in"` |
| 1311 | Out int64 `json:"out"` |
| 1312 | Rejected int64 `json:"rejected"` |
| 1313 | Allocs int `json:"allocs"` |
| 1314 | } |
| 1315 | |
| 1316 | type MeterSample struct { |
| 1317 | At time.Time `json:"at"` |
| 1318 | Apps map[string]MeterApp `json:"apps"` |
| 1319 | } |
| 1320 | |
| 1321 | type MeterChunk struct { |
| 1322 | V int `json:"v"` |
| 1323 | Box string `json:"box"` |
| 1324 | Hour string `json:"hour"` |
| 1325 | Samples []MeterSample `json:"samples"` |
| 1326 | } |
| 1327 | |
| 1328 | type Meter struct { |
| 1329 | mu sync.Mutex |
| 1330 | box string |
| 1331 | now func() time.Time |
| 1332 | cur map[Identity]*MeterApp |
| 1333 | } |
| 1334 | |
| 1335 | func NewMeter(box string, now func() time.Time) *Meter { |
| 1336 | return &Meter{box: box, now: now, cur: map[Identity]*MeterApp{}} |
| 1337 | } |
| 1338 | |
| 1339 | func (m *Meter) get(id Identity) *MeterApp { |
| 1340 | a, ok := m.cur[id] |
| 1341 | if !ok { |
| 1342 | a = &MeterApp{} |
| 1343 | m.cur[id] = a |
| 1344 | } |
| 1345 | return a |
| 1346 | } |
| 1347 | |
| 1348 | func (m *Meter) Add(id Identity, in, out, rejected int64) { |
| 1349 | m.mu.Lock() |
| 1350 | a := m.get(id) |
| 1351 | a.In += in; a.Out += out; a.Rejected += rejected |
| 1352 | m.mu.Unlock() |
| 1353 | } |
| 1354 | |
| 1355 | func (m *Meter) SetAllocs(id Identity, n int) { |
| 1356 | m.mu.Lock() |
| 1357 | m.get(id).Allocs = n |
| 1358 | m.mu.Unlock() |
| 1359 | } |
| 1360 | |
| 1361 | // Flush writes one sample of deltas since the previous flush, appended to |
| 1362 | // this box's hourly object by read-append-CAS. The relay is the only |
| 1363 | // writer of these objects, so the CAS only ever races with itself. |
| 1364 | func (m *Meter) Flush(ctx context.Context, s release.Store) error { |
| 1365 | m.mu.Lock() |
| 1366 | now := m.now() |
| 1367 | sample := MeterSample{At: now, Apps: map[string]MeterApp{}} |
| 1368 | for id, a := range m.cur { |
| 1369 | sample.Apps[id.Account+"/"+id.App] = *a |
| 1370 | m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept |
| 1371 | } |
| 1372 | m.mu.Unlock() |
| 1373 | hour := metrics.HourOf(now) |
| 1374 | key := MeterKey(m.box, hour) |
| 1375 | var lastErr error |
| 1376 | for attempt := 0; attempt < 3; attempt++ { |
| 1377 | body, etag, err := s.Get(ctx, key) |
| 1378 | var c MeterChunk |
| 1379 | switch { |
| 1380 | case errors.Is(err, os.ErrNotExist): |
| 1381 | c = MeterChunk{V: 1, Box: m.box, Hour: hour} |
| 1382 | etag = "" |
| 1383 | case err != nil: |
| 1384 | return err |
| 1385 | default: |
| 1386 | if json.Unmarshal(body, &c) != nil { |
| 1387 | c = MeterChunk{V: 1, Box: m.box, Hour: hour} |
| 1388 | } |
| 1389 | } |
| 1390 | c.Samples = append(c.Samples, sample) |
| 1391 | out, _ := json.Marshal(c) |
| 1392 | if _, err = s.Put(ctx, key, out, etag); err == nil { |
| 1393 | return nil |
| 1394 | } else if !errors.Is(err, release.ErrCASConflict) { |
| 1395 | return err |
| 1396 | } else { |
| 1397 | lastErr = err |
| 1398 | } |
| 1399 | } |
| 1400 | return lastErr |
| 1401 | } |
| 1402 | |
| 1403 | // LoadWindow returns every sample from the previous and current hour's |
| 1404 | // objects, for rebuilding rolling-hour budgets after a restart. |
| 1405 | func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time) ([]MeterSample, error) { |
| 1406 | var out []MeterSample |
| 1407 | for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} { |
| 1408 | body, _, err := s.Get(ctx, MeterKey(box, h)) |
| 1409 | if errors.Is(err, os.ErrNotExist) { |
| 1410 | continue |
| 1411 | } |
| 1412 | if err != nil { |
| 1413 | return nil, err |
| 1414 | } |
| 1415 | var c MeterChunk |
| 1416 | if json.Unmarshal(body, &c) == nil { |
| 1417 | out = append(out, c.Samples...) |
| 1418 | } |
| 1419 | } |
| 1420 | return out, nil |
| 1421 | } |
| 1422 | ``` |
| 1423 | |
| 1424 | - [ ] **Step 3: Run tests, commit** |
| 1425 | |
| 1426 | ```bash |
| 1427 | git add internal/relay/meter.go internal/relay/meter_test.go |
| 1428 | git commit -m "relay: per-app meter with its own hourly objects (spec §3 metering)" |
| 1429 | ``` |
| 1430 | |
| 1431 | --- |
| 1432 | |
| 1433 | ### Task 8: `Serve`, the unit, bootstrap arming, and the adopt restart |
| 1434 | |
| 1435 | **Files:** |
| 1436 | - Create: `internal/relay/serve.go` |
| 1437 | - Create: `internal/host/units/carlos-relay.service` |
| 1438 | - Modify: `internal/host/bootstrap.go:72` (`enabledUnits`), `:96-110` (`systemctlPlan`) |
| 1439 | - Modify: `internal/host/systemapply.go:33` (unit const), `:382` (restart), `:568` (`restartEdge`) |
| 1440 | - Modify: `internal/host/host_test.go:29` (`expectedUnits`) |
| 1441 | - Modify: `relay.go` (root; created here with `serve` and `keygen`, other verbs in Task 11), `main.go:76` (dispatch) |
| 1442 | - Test: `internal/host/host_test.go`, `internal/relay/serve_test.go` |
| 1443 | |
| 1444 | **Interfaces:** |
| 1445 | - Produces: |
| 1446 | ```go |
| 1447 | // internal/relay |
| 1448 | type Config struct { |
| 1449 | Box, Region string; MastersPath string; PublicIP net.IP; DenyCIDRs []*net.IPNet |
| 1450 | Store release.Store; Limits Limits; PressurePath string /* /proc/pressure/memory */ |
| 1451 | } |
| 1452 | func ConfigFromEnv(ctx context.Context, getenv func(string) string, imdsIP, imdsRegion func() string, store release.Store) (Config, error) |
| 1453 | func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error |
| 1454 | ``` |
| 1455 | - Env read by `ConfigFromEnv`: `CARLOS_RELAY_PUBLIC_IP` (override), `CARLOS_RELAY_DENY_CIDRS` (comma list), `CARLOS_AWS_REGION` (override), `CARLOS_RELAY_MASTERS_PATH` (default `/etc/carlos/relay-masters`), `CARLOS_RELAY_LIMIT_*`. |
| 1456 | |
| 1457 | - [ ] **Step 1: The unit file** |
| 1458 | |
| 1459 | ```ini |
| 1460 | # The platform STUN/TURN relay (media roadmap stage 4a). Armed only on a |
| 1461 | # box that carries /etc/carlos/relay-masters; ConditionPathExists keeps a |
| 1462 | # box without a master from ever listening. Shares the box with the edge and |
| 1463 | # must lose first: bounded memory, PSI-driven shedding in the process, and |
| 1464 | # the highest OOM adjustment there is. |
| 1465 | [Unit] |
| 1466 | Description=carlos relay (STUN and TURN) |
| 1467 | After=network-online.target |
| 1468 | Wants=network-online.target |
| 1469 | ConditionPathExists=/etc/carlos/relay-masters |
| 1470 | |
| 1471 | [Service] |
| 1472 | EnvironmentFile=-/etc/carlos/host.env |
| 1473 | ExecStart=/opt/carlos/carlos relay serve |
| 1474 | Restart=always |
| 1475 | RestartSec=1 |
| 1476 | |
| 1477 | User=carlos |
| 1478 | Group=carlos |
| 1479 | NoNewPrivileges=yes |
| 1480 | ProtectSystem=strict |
| 1481 | ProtectHome=yes |
| 1482 | ReadWritePaths=/data/relay /run/carlos |
| 1483 | PrivateTmp=yes |
| 1484 | PrivateDevices=yes |
| 1485 | ProtectClock=yes |
| 1486 | ProtectHostname=yes |
| 1487 | ProtectKernelTunables=yes |
| 1488 | ProtectKernelModules=yes |
| 1489 | ProtectKernelLogs=yes |
| 1490 | ProtectControlGroups=yes |
| 1491 | ProtectProc=invisible |
| 1492 | RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX |
| 1493 | RestrictNamespaces=yes |
| 1494 | RestrictRealtime=yes |
| 1495 | RestrictSUIDSGID=yes |
| 1496 | LockPersonality=yes |
| 1497 | MemoryDenyWriteExecute=yes |
| 1498 | SystemCallFilter=@system-service |
| 1499 | SystemCallErrorNumber=EPERM |
| 1500 | SystemCallArchitectures=native |
| 1501 | MemoryHigh=160M |
| 1502 | MemoryMax=256M |
| 1503 | OOMScoreAdjust=1000 |
| 1504 | TasksMax=512 |
| 1505 | Nice=5 |
| 1506 | CPUWeight=50 |
| 1507 | |
| 1508 | [Install] |
| 1509 | WantedBy=multi-user.target |
| 1510 | ``` |
| 1511 | |
| 1512 | - [ ] **Step 2: Arm it in bootstrap** (`internal/host/bootstrap.go`) |
| 1513 | |
| 1514 | Change line 72: |
| 1515 | ```go |
| 1516 | var enabledUnits = []string{"carlos-edge.service", "carlos-relay.service", "carlos-adopt.timer", "carlos-adopt.path", "carlos-stage.path", "carlos-restart.path"} |
| 1517 | ``` |
| 1518 | Change `systemctlPlan` so the relay is started, not only enabled (timers and paths already are): |
| 1519 | ```go |
| 1520 | for _, u := range units { |
| 1521 | if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") || u == "carlos-relay.service" { |
| 1522 | armed = append(armed, u) |
| 1523 | } |
| 1524 | } |
| 1525 | ``` |
| 1526 | `ConditionPathExists` makes `start` a harmless no-op on a box with no masters file, so this is safe fleet-wide. Also create `/data/relay` in `bootstrap()` beside the other data dirs (find where `/data` subdirectories are created and add `relay` with the same owner and mode). |
| 1527 | |
| 1528 | - [ ] **Step 3: Restart on platform update** (`internal/host/systemapply.go`) |
| 1529 | |
| 1530 | Add beside line 33: `systemRelayUnit = "carlos-relay"`. Add after `restartEdge`: |
| 1531 | ```go |
| 1532 | // restartRelay mirrors restartEdge for the relay; a unit that is inactive |
| 1533 | // because its ConditionPathExists failed returns an error from restart, |
| 1534 | // which is not a failure of the update, so it is logged and swallowed. |
| 1535 | func (a *systemApplier) restartRelay() { |
| 1536 | a.ops.systemctl("reset-failed", systemRelayUnit) |
| 1537 | if _, err := a.ops.systemctl("try-restart", systemRelayUnit); err != nil { |
| 1538 | log.Printf("system: relay try-restart: %v", err) |
| 1539 | } |
| 1540 | } |
| 1541 | ``` |
| 1542 | Call `a.restartRelay()` immediately after each `a.restartEdge()` succeeds (lines 382 and the rollback site near 692). |
| 1543 | |
| 1544 | - [ ] **Step 4: Tests** in `internal/host/host_test.go` |
| 1545 | |
| 1546 | Add `"carlos-relay.service"` to `expectedUnits` (keep the slice sorted). Add: |
| 1547 | ```go |
| 1548 | func TestBootstrapStartsRelayUnit(t *testing.T) { |
| 1549 | plan := systemctlPlan(enabledUnits) |
| 1550 | var started []string |
| 1551 | for _, step := range plan { |
| 1552 | if step[0] == "start" { |
| 1553 | started = step[1:] |
| 1554 | } |
| 1555 | } |
| 1556 | if !slices.Contains(started, "carlos-relay.service") { |
| 1557 | t.Fatalf("relay not started by plan: %v", plan) |
| 1558 | } |
| 1559 | if slices.Contains(started, "carlos-edge.service") { |
| 1560 | t.Fatalf("edge must stay start-by-hand: %v", plan) |
| 1561 | } |
| 1562 | } |
| 1563 | ``` |
| 1564 | |
| 1565 | - [ ] **Step 5: `Serve` and `ConfigFromEnv`** |
| 1566 | |
| 1567 | ```go |
| 1568 | // Serve runs the relay until ctx is done: watches the masters file, reads |
| 1569 | // authority every minute (and on control/serial change), heartbeats, flushes |
| 1570 | // the meter, ticks expiry, and sheds under memory pressure. |
| 1571 | func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error { |
| 1572 | var ( |
| 1573 | mu sync.Mutex |
| 1574 | masters *Masters |
| 1575 | mastersE string |
| 1576 | auth *Authority |
| 1577 | lease Lease |
| 1578 | ) |
| 1579 | now := time.Now |
| 1580 | meter := NewMeter(cfg.Box, now) |
| 1581 | if samples, err := LoadWindow(ctx, cfg.Store, cfg.Box, now()); err == nil { |
| 1582 | // seeded into Admission below after NewServer |
| 1583 | _ = samples |
| 1584 | } |
| 1585 | srv, err := NewServer(ServerConfig{ |
| 1586 | Region: cfg.Region, RelayIP: cfg.PublicIP, UDP: ":3478", TCP: ":3478", MinPort: 49152, MaxPort: 65535, |
| 1587 | DenyCIDRs: cfg.DenyCIDRs, Limits: cfg.Limits, Now: now, Log: logf, Meter: meter, |
| 1588 | Masters: func() *Masters { mu.Lock(); defer mu.Unlock(); return masters }, |
| 1589 | Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth }, |
| 1590 | Lease: func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) }, |
| 1591 | }) |
| 1592 | if err != nil { |
| 1593 | return err |
| 1594 | } |
| 1595 | defer srv.Close() |
| 1596 | // (seed: for each sample in the window, srv.admission.Seed(id, sample.At, app.In+app.Out)) |
| 1597 | |
| 1598 | go WatchMasters(ctx, cfg.MastersPath, func(m *Masters, err error) { |
| 1599 | mu.Lock() |
| 1600 | if err != nil { mastersE = err.Error() } else { masters, mastersE = m, "" } |
| 1601 | mu.Unlock() |
| 1602 | if err == nil { srv.tick() } // retire_at may have passed |
| 1603 | }) |
| 1604 | |
| 1605 | readAuth := func() { |
| 1606 | a, _, err := ReadAuthority(ctx, cfg.Store) |
| 1607 | t := now() |
| 1608 | mu.Lock() |
| 1609 | if err == nil { auth = a; lease.Renew(t) } else { logf("relay: authority read: %v", err) } |
| 1610 | live := lease.Live(t) |
| 1611 | hb := Heartbeat{Box: cfg.Box, Region: cfg.Region, ReadAt: t, Listeners: srv.Listeners(), Kids: kidsOf(masters), MastersError: mastersE} |
| 1612 | if auth != nil { hb.WrittenAt = auth.WrittenAt } |
| 1613 | mu.Unlock() |
| 1614 | if !live { srv.EndAll("lease lapsed") } |
| 1615 | srv.EndWhere(func(b *Binding) bool { // authority changed under it |
| 1616 | x, ok := auth.Lookup(b.ID.Account, b.ID.App) |
| 1617 | return !ok || x.Incarnation != b.Incarnation || !slices.Contains(x.Regions, cfg.Region) |
| 1618 | }, "authority") |
| 1619 | if err := WriteHeartbeat(ctx, cfg.Store, hb); err != nil { logf("relay: heartbeat: %v", err) } |
| 1620 | } |
| 1621 | readAuth() |
| 1622 | authTick := time.NewTicker(time.Minute) |
| 1623 | serialTick := time.NewTicker(3 * time.Second) |
| 1624 | meterTick := time.NewTicker(time.Minute) |
| 1625 | expiryTick := time.NewTicker(10 * time.Second) |
| 1626 | psiTick := time.NewTicker(5 * time.Second) |
| 1627 | defer func() { for _, t := range []*time.Ticker{authTick, serialTick, meterTick, expiryTick, psiTick} { t.Stop() } }() |
| 1628 | var lastSerial int64 = -1 |
| 1629 | for { |
| 1630 | select { |
| 1631 | case <-ctx.Done(): |
| 1632 | return nil |
| 1633 | case <-authTick.C: |
| 1634 | readAuth() |
| 1635 | case <-serialTick.C: |
| 1636 | if doc, _, err := release.ReadSerial(ctx, cfg.Store); err == nil && doc.Serial != lastSerial { |
| 1637 | lastSerial = doc.Serial |
| 1638 | readAuth() |
| 1639 | } |
| 1640 | case <-meterTick.C: |
| 1641 | if err := meter.Flush(ctx, cfg.Store); err != nil { logf("relay: meter flush: %v", err) } |
| 1642 | case <-expiryTick.C: |
| 1643 | srv.tick() |
| 1644 | case <-psiTick.C: |
| 1645 | if some, err := readPSI(cfg.PressurePath); err == nil { |
| 1646 | if some > 20 { srv.ShedOldest(1) } // one per tick until under 10; ShedOldest is a Server method closing the oldest binding |
| 1647 | } |
| 1648 | } |
| 1649 | } |
| 1650 | } |
| 1651 | ``` |
| 1652 | `readPSI` parses the `some avg10=` field of `/proc/pressure/memory`. `Binding` gains an `Incarnation string` field set at `created` from the parsed username (add it in Task 6 if not already present). Write `ConfigFromEnv` reading the env names above, using `imdsIP()`/`imdsRegion()` when the overrides are empty, and refusing to start with no public IP. |
| 1653 | |
| 1654 | - [ ] **Step 6: The verb** |
| 1655 | |
| 1656 | Create `relay.go` at the root with `runRelay(args)` dispatching `serve` and `keygen` now (other sub-verbs in Task 11): |
| 1657 | ```go |
| 1658 | func runRelayServe(args []string) { |
| 1659 | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) |
| 1660 | defer stop() |
| 1661 | store, err := openDeployStoreForCLI(ctx) // see below |
| 1662 | if err != nil { log.Fatalf("relay serve: %v", err) } |
| 1663 | cfg, err := relay.ConfigFromEnv(ctx, os.Getenv, host.ReadPublicIP, host.ReadPlacementRegion, store) |
| 1664 | if err != nil { log.Fatalf("relay serve: %v", err) } |
| 1665 | if err := relay.Serve(ctx, cfg, log.Printf); err != nil { log.Fatalf("relay serve: %v", err) } |
| 1666 | } |
| 1667 | |
| 1668 | func runRelayKeygen(_ []string) { |
| 1669 | key := make([]byte, 32) |
| 1670 | rand.Read(key) |
| 1671 | kid := make([]byte, 4) |
| 1672 | rand.Read(kid) |
| 1673 | fmt.Printf("%s %s -\n", hex.EncodeToString(kid), base64.StdEncoding.EncodeToString(key)) |
| 1674 | } |
| 1675 | ``` |
| 1676 | Export `host.ReadPublicIP` and `host.ReadPlacementRegion` as thin wrappers over the existing unexported `readPublicIP`/`readPlacementRegion` (`internal/host/agent.go:1808`, `:1831`), and `host.OpenDeployStore(ctx)` over `openDeployStore` (`agent.go:2692`); the box label comes from `host.BoxLabel()` wrapping `metricsBoxLabel(os.Getenv, readMachineID())`. Add `case "relay": runRelay(args)` to `main.go` after `store`, and a line to `usage()`. |
| 1677 | |
| 1678 | - [ ] **Step 7: Run all tests, cross-compile, commit** |
| 1679 | |
| 1680 | Run: `go test ./internal/relay/ ./internal/host/ -count=1 && go build ./... && for t in darwin/arm64 windows/amd64 linux/arm64; do GOOS=${t%/*} GOARCH=${t#*/} CGO_ENABLED=0 go build ./... || echo FAIL $t; done` |
| 1681 | |
| 1682 | ```bash |
| 1683 | git add internal/relay/serve.go internal/host/units/carlos-relay.service internal/host/bootstrap.go internal/host/systemapply.go internal/host/host_test.go internal/host/agent.go internal/host/metrics_loop.go relay.go main.go |
| 1684 | git commit -m "relay: carlos relay serve, the unit, bootstrap arming and adopt restart (spec §3)" |
| 1685 | ``` |
| 1686 | |
| 1687 | --- |
| 1688 | |
| 1689 | ### Task 9: The per-app record and the config value |
| 1690 | |
| 1691 | **Files:** |
| 1692 | - Create: `internal/relay/record.go`, `internal/relay/config.go` |
| 1693 | - Test: `internal/relay/record_test.go`, `internal/relay/config_test.go` |
| 1694 | |
| 1695 | **Interfaces:** |
| 1696 | - Produces: |
| 1697 | ```go |
| 1698 | func RecordKey(acct, app string) string // "apps/<acct>/<app>/relay.json" |
| 1699 | type RouteCutover struct{ Host string; Suffixes map[string]string /* region→suffix */; At time.Time; By string } |
| 1700 | type Record struct { |
| 1701 | Enabled bool; Incarnation string; Regions []string |
| 1702 | Op int64; Phase string /* "config"|"aggregate"|"done" */ |
| 1703 | IssuedSuffixes map[string]string /* region→username_suffix last issued */ |
| 1704 | Cutovers []RouteCutover /* member acks for TCP exec routes */ |
| 1705 | EnabledAt, DisabledAt time.Time |
| 1706 | } |
| 1707 | func GetRecord(ctx, s release.Store, acct, app string) (*Record, string, error) |
| 1708 | func PutRecord(ctx, s release.Store, acct, app string, r *Record, etag string) (string, error) |
| 1709 | func NewIncarnation() string // 16 lowercase hex from crypto/rand |
| 1710 | type RegionMaster struct{ Region, Kid string; Master []byte } |
| 1711 | type RegionURLs func(region string) []string // urls for a region, from the platform domain |
| 1712 | func BuildConfig(acct, app string, r *Record, masters []RegionMaster, urls RegionURLs) (value string, suffixes map[string]string, err error) |
| 1713 | // value is base64(std, padded) of the JSON array in spec §2, elements in r.Regions order; masters must cover every region in r.Regions. |
| 1714 | func ParseConfig(value string) ([]ConfigEntry, error) // for the config-ack (Task 12) and tests |
| 1715 | type ConfigEntry struct{ Region string; URLs []string; Key string; UsernameSuffix string } |
| 1716 | ``` |
| 1717 | |
| 1718 | - [ ] **Step 1: Tests** |
| 1719 | |
| 1720 | ```go |
| 1721 | func TestBuildConfigMatchesContract(t *testing.T) { |
| 1722 | m := sha256.Sum256([]byte("carlos-relay-test-master")) |
| 1723 | r := &Record{Enabled: true, Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}} |
| 1724 | urls := func(region string) []string { |
| 1725 | h := "relay." + region + ".oncarlos.com:3478" |
| 1726 | return []string{"stun:" + h, "turn:" + h + "?transport=udp", "turn:" + h + "?transport=tcp"} |
| 1727 | } |
| 1728 | val, suffixes, err := BuildConfig("abc", "meet", r, []RegionMaster{{Region: "eu-west-1", Kid: "k1", Master: m[:]}}, urls) |
| 1729 | if err != nil { t.Fatal(err) } |
| 1730 | if err := appconfig.ValidVarValue(val); err != nil { t.Fatalf("not env-safe: %v", err) } |
| 1731 | entries, err := ParseConfig(val) |
| 1732 | if err != nil || len(entries) != 1 { t.Fatalf("parse = %+v %v", entries, err) } |
| 1733 | e := entries[0] |
| 1734 | if e.Key != "uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=" || e.UsernameSuffix != "abc:meet:0123456789abcdef:k1" || suffixes["eu-west-1"] != e.UsernameSuffix { |
| 1735 | t.Fatalf("entry = %+v", e) |
| 1736 | } |
| 1737 | if len(e.URLs) != 3 || e.URLs[0] != "stun:relay.eu-west-1.oncarlos.com:3478" { t.Fatalf("urls = %v", e.URLs) } |
| 1738 | } |
| 1739 | |
| 1740 | func TestBuildConfigRefusesMissingMaster(t *testing.T) { |
| 1741 | r := &Record{Enabled: true, Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1", "us-east-1"}} |
| 1742 | if _, _, err := BuildConfig("abc", "meet", r, []RegionMaster{{Region: "eu-west-1", Kid: "k1", Master: make([]byte, 32)}}, func(string) []string { return nil }); err == nil { |
| 1743 | t.Fatal("missing us-east-1 master accepted") |
| 1744 | } |
| 1745 | } |
| 1746 | |
| 1747 | func TestNewIncarnation(t *testing.T) { |
| 1748 | a, b := NewIncarnation(), NewIncarnation() |
| 1749 | if !incarnationRe.MatchString(a) || a == b { t.Fatalf("%q %q", a, b) } |
| 1750 | } |
| 1751 | |
| 1752 | func TestRecordRoundTrip(t *testing.T) { |
| 1753 | s := newMemStore() |
| 1754 | ctx := context.Background() |
| 1755 | r, etag, err := GetRecord(ctx, s, "abc", "meet") |
| 1756 | if err != nil || r != nil || etag != "" { t.Fatalf("absent = %v %q %v", r, etag, err) } |
| 1757 | etag, err = PutRecord(ctx, s, "abc", "meet", &Record{Enabled: true, Incarnation: NewIncarnation(), Op: 1, Phase: "config"}, "") |
| 1758 | if err != nil { t.Fatal(err) } |
| 1759 | if _, err := PutRecord(ctx, s, "abc", "meet", &Record{}, "stale-etag"); !errors.Is(err, release.ErrCASConflict) { |
| 1760 | t.Fatalf("stale put = %v, want CAS conflict", err) |
| 1761 | } |
| 1762 | r, _, _ = GetRecord(ctx, s, "abc", "meet") |
| 1763 | if !r.Enabled || r.Op != 1 { t.Fatalf("record = %+v", r) } |
| 1764 | } |
| 1765 | ``` |
| 1766 | |
| 1767 | - [ ] **Step 2: Implement** `record.go` (mirror `internal/objstore/objstore.go:132-170` Get/Put with `os.ErrNotExist` → nil) and `config.go`: |
| 1768 | |
| 1769 | ```go |
| 1770 | type ConfigEntry struct { |
| 1771 | Region string `json:"region"` |
| 1772 | URLs []string `json:"urls"` |
| 1773 | Key string `json:"key"` |
| 1774 | UsernameSuffix string `json:"username_suffix"` |
| 1775 | } |
| 1776 | |
| 1777 | func BuildConfig(acct, app string, r *Record, masters []RegionMaster, urls RegionURLs) (string, map[string]string, error) { |
| 1778 | byRegion := map[string]RegionMaster{} |
| 1779 | for _, m := range masters { byRegion[m.Region] = m } |
| 1780 | entries := make([]ConfigEntry, 0, len(r.Regions)) |
| 1781 | suffixes := map[string]string{} |
| 1782 | for _, region := range r.Regions { |
| 1783 | m, ok := byRegion[region] |
| 1784 | if !ok { return "", nil, fmt.Errorf("relay: no issuer master for region %s", region) } |
| 1785 | key := DeriveKey(m.Master, acct, app, r.Incarnation) |
| 1786 | suffix := Suffix(acct, app, r.Incarnation, m.Kid) |
| 1787 | suffixes[region] = suffix |
| 1788 | entries = append(entries, ConfigEntry{Region: region, URLs: urls(region), Key: base64.StdEncoding.EncodeToString(key), UsernameSuffix: suffix}) |
| 1789 | } |
| 1790 | js, err := json.Marshal(entries) |
| 1791 | if err != nil { return "", nil, err } |
| 1792 | return base64.StdEncoding.EncodeToString(js), suffixes, nil |
| 1793 | } |
| 1794 | |
| 1795 | func ParseConfig(value string) ([]ConfigEntry, error) { |
| 1796 | js, err := base64.StdEncoding.DecodeString(value) |
| 1797 | if err != nil { return nil, fmt.Errorf("relay: CARLOS_RELAY is not base64: %w", err) } |
| 1798 | var out []ConfigEntry |
| 1799 | if err := json.Unmarshal(js, &out); err != nil { return nil, fmt.Errorf("relay: CARLOS_RELAY is not a JSON array: %w", err) } |
| 1800 | return out, nil |
| 1801 | } |
| 1802 | ``` |
| 1803 | |
| 1804 | - [ ] **Step 3: Run tests, commit** |
| 1805 | |
| 1806 | ```bash |
| 1807 | git add internal/relay/record.go internal/relay/config.go internal/relay/*_test.go |
| 1808 | git commit -m "relay: per-app record and the CARLOS_RELAY value (spec §2, §5)" |
| 1809 | ``` |
| 1810 | |
| 1811 | --- |
| 1812 | |
| 1813 | ### Task 10: Console operations: writer lease, aggregate publish, reconcile (Opus) |
| 1814 | |
| 1815 | **Files:** |
| 1816 | - Create: `internal/console/relayops.go` |
| 1817 | - Test: `internal/console/relayops_test.go` |
| 1818 | |
| 1819 | **Interfaces:** |
| 1820 | - Consumes: Task 9; `appconfig.Apply` (`internal/appconfig/generation.go:119`), `appconfig.Seal`, `release.BumpSerial`, `s.cfg.Releases`, `s.cfg.SecretsPub`. |
| 1821 | - Produces (methods on `*Server`, all requiring the writer lease): |
| 1822 | ```go |
| 1823 | const relayWriterKey = "control/relay-writer.json" |
| 1824 | const relayIssuerKey = "control/relay-issuer.json" // {"<region>":"<kid>"} |
| 1825 | type relayWriter struct{ /* holder id, expiry; renew every 20s; 60s TTL */ } |
| 1826 | func (s *Server) relayWriterLoop(ctx) // acquires/renews; sets s.relayHeld atomically |
| 1827 | func (s *Server) relayHold() bool |
| 1828 | func (s *Server) relayMasters(region string) (kid string, master []byte, err error) // console secret CARLOS_RELAY_MASTERS_<REGION> + issuer pointer |
| 1829 | func (s *Server) relayEnable(ctx, acct *Account, app string, regions []string, by string) (*relay.Record, error) |
| 1830 | func (s *Server) relayDisable(ctx, acct *Account, app string, by string) (*relay.Record, error) |
| 1831 | func (s *Server) relayAdopted(ctx, acct *Account, app, host string, by string) (*relay.Record, error) |
| 1832 | func (s *Server) relayRotate(ctx, region string) error // gate on heartbeats, move issuer pointer, re-issue every enabled app in region |
| 1833 | func (s *Server) relayPublishAggregate(ctx) error // etag-first rebuild from a listing, CAS, retry |
| 1834 | func (s *Server) relayReconcile(ctx) error // every 5 min and at start; converges config and aggregate from every record |
| 1835 | ``` |
| 1836 | - Console config additions (`cmd/console/main.go` beside `storeOperators()`): `CARLOS_RELAY_REGIONS` (comma list, ordered) → `Config.RelayRegions []string`; `CARLOS_RELAY_MASTERS_<REGION>` read at use time through the console's own sealed-secret accessor (find how the console reads its own sealed secrets, `grep -rn "SecretsPriv\|consoleSecret" internal/console cmd/console`); `Config.PlatformDomain` for relay hostnames (`relay.<region>.<platform domain>`). |
| 1837 | |
| 1838 | - [ ] **Step 1: Tests** (each with the console fixture; `fx.rel` is the `release.Store`) |
| 1839 | |
| 1840 | ```go |
| 1841 | func TestRelayEnableWritesRecordConfigAndAggregate(t *testing.T) { |
| 1842 | fx, _ := relayFixture(t, "code-relay-enable") // like storeFixture plus RelayRegions=["eu-west-1"], a test master in the fixture's console secrets, SecretsPub set |
| 1843 | ctx := context.Background() |
| 1844 | rec, err := fx.srv.relayEnable(ctx, fx.acct(), "titogo", nil, "alice@keymail.dev") |
| 1845 | if err != nil { t.Fatal(err) } |
| 1846 | if !rec.Enabled || rec.Phase != "done" || rec.Op != 1 || len(rec.Incarnation) != 16 { t.Fatalf("record = %+v", rec) } |
| 1847 | m, _, _ := appconfig.Read(ctx, fx.rel, fx.sqid, "titogo") // whatever reads the default bundle manifest |
| 1848 | if v, ok := m.Vars["CARLOS_RELAY"]; !ok || v.Kind != "secret" { t.Fatalf("CARLOS_RELAY = %+v", v) } |
| 1849 | a, _, _ := relay.ReadAuthority(ctx, fx.rel) |
| 1850 | if x, ok := a.Lookup(fx.sqid, "titogo"); !ok || x.Incarnation != rec.Incarnation { t.Fatalf("aggregate = %+v", a) } |
| 1851 | } |
| 1852 | |
| 1853 | func TestRelayRepeatEnableKeepsIncarnationAndWritesNoConfig(t *testing.T) { /* enable twice; second returns same Incarnation, Op unchanged, config generation unchanged */ } |
| 1854 | |
| 1855 | func TestRelayDisableThenEnableIsANewIncarnation(t *testing.T) { /* enable, disable (aggregate lacks app, CARLOS_RELAY unset), enable: Incarnation differs */ } |
| 1856 | |
| 1857 | func TestRelayPublishAggregateIsEtagFirst(t *testing.T) { |
| 1858 | // Seed two enabled records A and B. Read the aggregate etag. Disable A directly via PutRecord + relayPublishAggregate. |
| 1859 | // Then call the internal publish helper with the STALE etag and a listing that still has A: it must return ErrCASConflict and, on retry, publish without A. |
| 1860 | } |
| 1861 | |
| 1862 | func TestRelayReconcileFinishesInterruptedEnable(t *testing.T) { |
| 1863 | // PutRecord{Enabled:true, Op:1, Phase:"config"} with no config and no aggregate; relayReconcile; expect CARLOS_RELAY set, aggregate has app, Phase "done". |
| 1864 | } |
| 1865 | |
| 1866 | func TestRelayReconcileUndoesStaleConfigAfterDisable(t *testing.T) { |
| 1867 | // Disabled record but CARLOS_RELAY present in config; relayReconcile; expect it unset. |
| 1868 | } |
| 1869 | |
| 1870 | func TestRelayReconcileRepairsDeletionWithoutRecord(t *testing.T) { |
| 1871 | // Aggregate lists an app with no record; relayReconcile; aggregate no longer lists it. |
| 1872 | } |
| 1873 | |
| 1874 | func TestRelayEnableRefusesRouteWithoutDelivery(t *testing.T) { |
| 1875 | // fx.addRoute(registry.Route{Host: "x.titogo...", App: "titogo", Addr: "10.0.0.5:8080"}) with no exec_delivery; relayEnable errors naming the host. |
| 1876 | } |
| 1877 | |
| 1878 | func TestRelayRotateRefusesUntilHeartbeatsCarryKid(t *testing.T) { |
| 1879 | // Two masters in the fixture secret for eu-west-1; issuer pointer k1; heartbeat for box1 lists only k1; relayRotate("eu-west-1") errors; write heartbeat listing k2; relayRotate succeeds; issuer pointer now k2; enabled app's IssuedSuffixes end in ":k2". |
| 1880 | } |
| 1881 | |
| 1882 | func TestRelayWriterLeaseExcludesSecondHolder(t *testing.T) { |
| 1883 | // Two Server values over one store; first acquires; second's acquire fails until the first's expiry passes on a fake clock. |
| 1884 | } |
| 1885 | ``` |
| 1886 | |
| 1887 | - [ ] **Step 2: Implement** in this order, each behind the lease: `relayWriterLoop` (CAS on `relayWriterKey` with `{holder, expires_at}`; acquire if absent or expired; renew every 20s; on failed renew set held=false and let in-flight ops fail their next `requireRelayLease()` check); `relayMasters` (parse `CARLOS_RELAY_MASTERS_<REGION>` as `kid:key[,kid:key]`, read `relayIssuerKey`, return the named one); `relayPublishAggregate` exactly per spec §5 (read aggregate etag first, then `ListDirs("apps/")` and each account's apps to find every `relay.json`, build from records with `Enabled`, `Put` with the first etag, on `ErrCASConflict` restart, four attempts); `relayApplyConfig(ctx, acct, app string, rec *relay.Record)` (build via `relay.BuildConfig` with `urls(region) = []string{"stun:relay."+region+"."+domain+":3478", "turn:...?transport=udp", "turn:...?transport=tcp"}`, `appconfig.Seal` the value with `s.cfg.SecretsPub`, `appconfig.Apply(ctx, s.cfg.Releases, acct.Sqid, app, map[string]appconfig.Var{"CARLOS_RELAY": {Kind: "secret", Envelope: env}}, nil)`; for disable `Apply(..., nil, []string{"CARLOS_RELAY"})`); `relayEnable`/`relayDisable` as the spec's transition table with a per-app `sync.Mutex` from `s.relayAppLocks sync.Map`; the route-delivery check using `s.registryRoutesForApp` (whatever the instances API uses to list an app's routes and their `Backing`/`Addr`) and `appconfig`'s exec-delivery resolution; `relayReconcile` per spec §5 (for every record: if enabled, expected config value equals current → skip, else re-apply; if disabled and present → unset; then unconditional `relayPublishAggregate`); `relayRotate` per spec §5 step 4. Start `relayWriterLoop` and a five-minute `relayReconcile` ticker from the console's existing background-loop startup, gated on `len(s.cfg.RelayRegions) > 0`. |
| 1888 | |
| 1889 | - [ ] **Step 3: Run tests, commit** |
| 1890 | |
| 1891 | ```bash |
| 1892 | git add internal/console/relayops.go internal/console/relayops_test.go cmd/console/main.go internal/console/server.go |
| 1893 | git commit -m "console: relay operations under a writer lease, etag-first aggregate, converging reconcile (spec §5)" |
| 1894 | ``` |
| 1895 | |
| 1896 | --- |
| 1897 | |
| 1898 | ### Task 11: Console doors and the CLI |
| 1899 | |
| 1900 | **Files:** |
| 1901 | - Create: `internal/console/apirelay.go` |
| 1902 | - Modify: `internal/console/authcli.go:700` (routes), `apimode.go` (client calls), `relay.go` (root; add enable/status/disable/adopted/rotate), `docs/site/cli.md` (`### carlos relay`) |
| 1903 | - Test: `internal/console/apirelay_test.go` |
| 1904 | |
| 1905 | **Interfaces:** |
| 1906 | - Routes (`s.cliRoute`, `svcrole.Allow(svcrole.Admin)` for all but status which allows read roles too, matching store): |
| 1907 | ``` |
| 1908 | POST /api/cli/apps/{account}/{app}/relay body {"regions":[...]} → relayEnable |
| 1909 | GET /api/cli/apps/{account}/{app}/relay → status |
| 1910 | POST /api/cli/apps/{account}/{app}/relay/disable → relayDisable |
| 1911 | POST /api/cli/apps/{account}/{app}/relay/adopted body {"host":"..."} → relayAdopted |
| 1912 | POST /api/cli/relay/rotate body {"region":"..."} → relayRotate (operator: CARLOS_RELAY_OPERATORS, checked after auth, byte-identical 404 on failure) |
| 1913 | ``` |
| 1914 | - Status reply shape: |
| 1915 | ```go |
| 1916 | type apiRelayReply struct { |
| 1917 | Enabled bool; Incarnation string; Regions []string; Op int64; Phase string |
| 1918 | Issued map[string]string // region → username_suffix |
| 1919 | Routes []apiRelayRoute // per route: Host, Shape, Delivery, AckedAt, Cutover |
| 1920 | Regions []apiRelayRegion // Region, HeartbeatAge string, Current bool, LeaseLive bool, Listeners, Kids []string, Stale bool |
| 1921 | DisablePending []string // regions not yet past the disable |
| 1922 | RetireNoEarlierThan map[string]string // region → RFC3339, when rotate is in progress |
| 1923 | } |
| 1924 | ``` |
| 1925 | |
| 1926 | - [ ] **Step 1: Door tests** through `rawReq`/`getJSON`/`postJSON` with `relayURL(fx, app) = fx.ts.URL + "/api/cli/apps/" + acct + "/" + app + "/relay"`: 404 byte-identical for a non-member and an unknown app (compare bodies with the store door's 404); refusal message when `RelayRegions` empty; enable → 200 with `enabled:true` and 16-hex incarnation; status after enable shows the region row from a seeded heartbeat with `stale:true` when the heartbeat is six minutes old; disable → `enabled:false`; adopted for an unknown host → 404 with message; rotate without operator → 404. |
| 1927 | |
| 1928 | - [ ] **Step 2: Implement doors** on the `handleAPIStoreGet` pattern (`internal/console/apistore.go:299`): `apiAccountApp` gate, `storeNoBucket` equivalent when `s.cfg.Releases == nil`, `jsonError` on failures, `writeJSON` on success. Status assembles the reply from `relay.GetRecord`, the heartbeats under `control/relay-ack/`, the config-acks under `control/config-ack/` (Task 12's shape), and the route list. |
| 1929 | |
| 1930 | - [ ] **Step 3: CLI** in `relay.go`: `runRelay` dispatch adds `enable|status|disable|adopted|rotate`; each mirrors `runStoreStatus` (`store.go:157`): flags via `appFlag`, `accountFlags`, `--console`; `requireRelayAPISession` copied from `requireStoreAPISession` with its text adapted; `apiSession.relayEnable/relayGet/relayDisable/relayAdopted/relayRotate` in `apimode.go` on the `storeGet` shape (`apimode.go:1592`); `printRelayStatus` prints the reply as: desired state line, issued kids per region, one line per route with its delivery state, one line per region with heartbeat age and lease, and the retire-no-earlier-than line when present. |
| 1931 | |
| 1932 | - [ ] **Step 4: Docs**: add to `docs/site/cli.md` under "Your apps", after `### carlos store`: |
| 1933 | |
| 1934 | ````markdown |
| 1935 | ### carlos relay |
| 1936 | |
| 1937 | A STUN and TURN relay the platform runs for your app's WebRTC calls, so a |
| 1938 | browser can find its public address and, when two networks cannot reach |
| 1939 | each other directly, relay through the deployment instead of a third party. |
| 1940 | |
| 1941 | ```sh |
| 1942 | carlos relay enable --app meet |
| 1943 | carlos relay status --app meet |
| 1944 | ``` |
| 1945 | |
| 1946 | `enable` turns it on for one app and delivers `CARLOS_RELAY` to the app's |
| 1947 | config: a base64-wrapped JSON list of regions, each with its ICE server URLs, |
| 1948 | a key and a username suffix. The app mints a TURN credential per call from |
| 1949 | the key (see the relay design spec, §2). `status` shows what was issued, |
| 1950 | which routes have picked it up, and the health of each region's relay, |
| 1951 | none of which needs box access. `disable` removes it and ends every |
| 1952 | allocation the app holds. `adopted --host <route>` is how a member tells the |
| 1953 | platform that a TCP-address route it restarted by hand now runs the new |
| 1954 | key. `rotate --region <r>` is the operator's verb after staging a new |
| 1955 | master; `keygen` prints one. `serve` is the unit's entry point and not a |
| 1956 | thing to run by hand. |
| 1957 | ```` |
| 1958 | |
| 1959 | - [ ] **Step 5: Run the docsite gate and all tests, commit** |
| 1960 | |
| 1961 | Run: `go test ./internal/docsite/ ./internal/console/ -count=1 && go build ./...` |
| 1962 | |
| 1963 | ```bash |
| 1964 | git add internal/console/apirelay.go internal/console/apirelay_test.go internal/console/authcli.go apimode.go relay.go docs/site/cli.md |
| 1965 | git commit -m "relay: console doors and carlos relay enable|status|disable|adopted|rotate (spec §5)" |
| 1966 | ``` |
| 1967 | |
| 1968 | --- |
| 1969 | |
| 1970 | ### Task 12: The per-route config acknowledgement |
| 1971 | |
| 1972 | **Files:** |
| 1973 | - Create: `internal/appconfig/configack.go` |
| 1974 | - Modify: `internal/appconfig/sync.go:331-434` (`syncHost` gains an ack writer callback), `internal/host/agent.go:2937` (pass the store and box) |
| 1975 | - Test: `internal/appconfig/configack_test.go`, `internal/appconfig/sync_test.go` |
| 1976 | |
| 1977 | **Interfaces:** |
| 1978 | - Produces: |
| 1979 | ```go |
| 1980 | func ConfigAckKey(box, host string) string // "control/config-ack/<box>/<host>.json" |
| 1981 | type ConfigAck struct { |
| 1982 | Box, Host, Shape string // shape: "unit" | "socket-exec" | "tcp-exec-file" |
| 1983 | At time.Time |
| 1984 | Restarted bool; RestartError string |
| 1985 | RelaySuffixes map[string]string // region → username_suffix parsed from CARLOS_RELAY, nil if absent |
| 1986 | } |
| 1987 | func WriteConfigAck(ctx, s release.Store, ack ConfigAck) error |
| 1988 | // Sync gains a parameter: ack func(ConfigAck) (nil = no acks) |
| 1989 | func Sync(ctx, reg, s, priv, envDir string, restart func(host string) error, ack func(ConfigAck)) error |
| 1990 | ``` |
| 1991 | |
| 1992 | - [ ] **Step 1: Tests**: `TestSyncWritesConfigAckNamingRelaySuffixes`: a route with a manifest carrying `CARLOS_RELAY` (built with `relay.BuildConfig` in the test) is synced with a recording `ack` func; the ack has `Shape: "unit"`, `Restarted: true`, `RelaySuffixes["eu-west-1"] == "abc:meet:...:k1"`. `TestSyncAckReportsRestartFailure`: restart returns an error; ack has `Restarted: false` and the error text. `TestSyncAckWithoutRelayHasNilSuffixes`. |
| 1993 | |
| 1994 | - [ ] **Step 2: Implement**: in `syncHost`, after the rename succeeds, build `ConfigAck{Host: rt.Host, Shape: shapeOf(rt), At: time.Now()}`; parse `CARLOS_RELAY` out of the rendered vars (the decrypted manifest is in scope; use `relay.ParseConfig` on its value if present and fill `RelaySuffixes`); call `restart`, set `Restarted`/`RestartError`; call `ack(a)` if non-nil. Import cycle check: `internal/appconfig` importing `internal/relay` is fine only if `internal/relay` does not import `appconfig`; it does not (Task 9's test imports `appconfig` but the package does not). `shapeOf`: `rt.Backing == "unit"` → `"unit"`; `rt.Socket != ""` → `"socket-exec"`; else `"tcp-exec-file"`. The exec-file path (`syncexec.go`) writes the same ack with `Restarted: false`. In `agent.go:2937`, pass `func(a appconfig.ConfigAck) { a.Box = box; if err := appconfig.WriteConfigAck(ctx, store, a); err != nil { log.Printf("adopt: config ack %s: %v", a.Host, err) } }`. |
| 1995 | |
| 1996 | - [ ] **Step 3: Run tests, commit** |
| 1997 | |
| 1998 | ```bash |
| 1999 | git add internal/appconfig/configack.go internal/appconfig/configack_test.go internal/appconfig/sync.go internal/appconfig/syncexec.go internal/appconfig/sync_test.go internal/host/agent.go |
| 2000 | git commit -m "appconfig: per-route config acknowledgement naming rendered relay identities (spec §5)" |
| 2001 | ``` |
| 2002 | |
| 2003 | --- |
| 2004 | |
| 2005 | ### Task 13: Infrastructure |
| 2006 | |
| 2007 | **Files:** |
| 2008 | - Modify: `infra/modules/region-host/variables.tf`, `main.tf`, `outputs.tf` |
| 2009 | - Modify (pinfra repo, separate PR there): `flagship/region-ie.tf` and siblings; `host-init.sh` if it stages `host.env` |
| 2010 | |
| 2011 | - [ ] **Step 1: Module**: add |
| 2012 | ```hcl |
| 2013 | variable "relay" { |
| 2014 | description = "Open the platform STUN/TURN relay's ports (udp+tcp 3478, udp 49152-65535). Off by default; a box also needs /etc/carlos/relay-masters before the unit starts." |
| 2015 | type = bool |
| 2016 | default = false |
| 2017 | } |
| 2018 | ``` |
| 2019 | and in `main.tf`, three more `dynamic "ingress"` entries driven by `var.relay ? [1] : []` with descriptions `"relay stun/turn udp"` (3478/3478/udp), `"relay stun/turn tcp"` (3478/3478/tcp), `"relay allocations"` (49152/65535/udp), all `0.0.0.0/0`. Add `output "vpc_cidr" { value = data.aws_vpc.selected.cidr_block }` (find the VPC data source the module already uses). Run `tofu validate` in the module directory. |
| 2020 | |
| 2021 | - [ ] **Step 2: Flagship** (pinfra repo): `relay = true` on the Ireland module first; an `aws_route53_record` `relay.eu-west-1.<platform domain>` A → `module.host_ie.eip_public_ip`; document in `flagship/README.md` that `/etc/carlos/relay-masters` and `CARLOS_RELAY_DENY_CIDRS=<vpc_cidr output>` are staged by hand like `host.env`, and that console config gets `CARLOS_RELAY_REGIONS=eu-west-1` and the sealed `CARLOS_RELAY_MASTERS_EU_WEST_1`. |
| 2022 | |
| 2023 | - [ ] **Step 3: Commit** (platform repo part) |
| 2024 | |
| 2025 | ```bash |
| 2026 | git add infra/modules/region-host/ |
| 2027 | git commit -m "infra: relay variable opens stun/turn ingress and outputs the VPC CIDR (spec §6)" |
| 2028 | ``` |
| 2029 | |
| 2030 | --- |
| 2031 | |
| 2032 | ### Task 14: Whole-branch gate and landing |
| 2033 | |
| 2034 | - [ ] **Step 1: Full gate**: `go build ./... && go vet ./... && go test ./... -count=1 && gofmt -l .` (empty output) and the three cross-compiles. |
| 2035 | - [ ] **Step 2: Spec coverage walk**: for each of spec §3's bullets, §4's steps, §5's transitions and §7's unit-test list, name the test that covers it; add any missing test. |
| 2036 | - [ ] **Step 3: Opus review** of the whole branch against the spec, with `superpowers:requesting-code-review`. |
| 2037 | - [ ] **Step 4: Land**: push to `origin`, `amadan branch describe carlos/platform relay-stun-turn-impl -body -`, wait for `amadan ci status -branch relay-stun-turn-impl` to pass, `amadan branch merge carlos/platform relay-stun-turn-impl -expect <full sha>`. |
| 2038 | - [ ] **Step 5: Flagship rollout** (operator, from the spec's §7 gate): stage masters and deny CIDRs on the Ireland box; apply pinfra; promote; confirm `carlos-relay.service` active; `carlos relay enable --app <meet app>`; run the §7 flagship checks in order; message the Meet session. |
| 2039 | |
| 2040 | --- |
| 2041 | |
| 2042 | ## Self-review |
| 2043 | |
| 2044 | **Spec coverage.** §2 contract → Tasks 1, 9 (vector test in 1, env-safety in 9). §3 process/unit → 8; hardening → 8; listeners and generator → 6; owner binding → 5, 6; termination → 6, 8; peer policy and EIP gate → 5, 6; limits → 4, 6; metering → 5, 6, 7; authority and lease → 3, 8. §4 masters → 2; console custody → 10; verification steps → 6. §5 records → 9; writer lease, transitions, publish, reconcile, rotate → 10; doors, status, adopted → 11; config-ack → 12. §6 → 13. §7 flagship gate → 14. §8 exclusions: none built. Gaps: none found. |
| 2045 | |
| 2046 | **Placeholders.** Task 6 step 3 and Task 10 step 2 are design-level for the Opus tasks by intent, with every function and behaviour named; Task 6's test file names the three tests. No "TBD". |
| 2047 | |
| 2048 | **Type consistency.** `Identity{Account, App}` used in Tasks 4 to 8; `Binding` gains `Incarnation` in Task 6 (noted in Task 8); `relay.Record` fields in Task 9 match Task 10 and 11 uses; `ConfigAck.RelaySuffixes` in Task 12 matches Task 11's status reading; `Sync` signature change in Task 12 must update the one caller at `agent.go:2937` and any test calling `Sync`. |
| 2049 | |
amadan