amadan

carlos / platform Public

plan: stage 4a relay implementation plan

Fourteen tasks from the accepted spec: internal/relay (keys, masters,
authority, limits, wrapped sockets, pion server, meter, serve), the unit
and bootstrap arming, the per-app record and CARLOS_RELAY value, console
operations under a writer lease, doors and CLI, the per-route config
ack, infra, and the landing gate.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 5965961946c0aee784d6fe9447ba5961d42adfeb parent 2923251
1 file changed, +2048 −0
  • docs/superpowers/plans/2026-09-07-relay-stun-turn.md +2048 −0

CI passed — run details

diff --git a/docs/superpowers/plans/2026-09-07-relay-stun-turn.md b/docs/superpowers/plans/2026-09-07-relay-stun-turn.md
new file mode 100644
index 000000000..2659de591
--- /dev/null
+++ b/docs/superpowers/plans/2026-09-07-relay-stun-turn.md
@@ -0,0 +1,2048 @@
+# Stage 4a relay (STUN and TURN) — Implementation Plan
+
+> **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.
+
+**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.
+
+**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.
+
+**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).
+
+**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`.
+
+## Global Constraints
+
+- **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.
+- **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.
+- **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.
+- **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`.)
+- **Config values must pass `appconfig.ValidVarValue`**: no quotes, backslashes, newlines or NUL. That is why `CARLOS_RELAY` is base64.
+- **`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.
+- **Every `carlos` command needs a `### carlos <verb>` heading in `docs/site/cli.md`** or `internal/docsite` fails the gate.
+- **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`.
+- **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`.
+- **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.
+
+---
+
+## File structure
+
+| Path | Responsibility |
+|---|---|
+| `internal/relay/keys.go` | HKDF derivation, credential minting, username parse; the §2 contract |
+| `internal/relay/masters.go` | masters file parse, `retire_at`, watched reload |
+| `internal/relay/authority.go` | `control/relay.json` shape, read with lease clock, heartbeat write |
+| `internal/relay/limits.go` | admission counters, rolling-hour budgets, token buckets |
+| `internal/relay/conn.go` | wrapped `net.PacketConn` and listener conns: counting, buckets, EIP port gate, close |
+| `internal/relay/server.go` | pion `ServerConfig` assembly: generator, auth handler, permission handler, bindings, termination timers |
+| `internal/relay/meter.go` | per-app counters, minute samples, hourly object read-append-CAS |
+| `internal/relay/serve.go` | `Serve(ctx, Config)`: wires the above, env parsing, PSI shedding loop |
+| `internal/relay/record.go` | `apps/<acct>/<app>/relay.json` shape, Get/Put, transitions |
+| `internal/relay/config.go` | build the `CARLOS_RELAY` value from a record and issuer masters |
+| `internal/relay/vector_test.go` | the §2 test vector |
+| `internal/host/units/carlos-relay.service` | the unit |
+| `internal/host/bootstrap.go` | arm and start the unit |
+| `internal/host/systemapply.go` | restart the relay on platform update |
+| `internal/appconfig/sync.go`, `internal/appconfig/configack.go` | per-route config acknowledgement |
+| `internal/console/apirelay.go` | doors: enable, status, disable, adopted, rotate |
+| `internal/console/relayops.go` | writer lease, operations, aggregate publish, reconcile |
+| `relay.go` (root) | `carlos relay` CLI: enable, status, disable, adopted, rotate, keygen, serve |
+| `apimode.go` | `apiSession.relay*` client calls |
+| `infra/modules/region-host/main.tf`, `variables.tf`, `outputs.tf` | `relay` variable, ingress, VPC CIDR output |
+| `docs/site/cli.md` | `### carlos relay` |
+
+---
+
+### Task 1: Key derivation and the credential contract
+
+**Files:**
+- Create: `internal/relay/keys.go`
+- Test: `internal/relay/vector_test.go`, `internal/relay/keys_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ const MaxCredentialLife = 4 * time.Hour
+ func DeriveKey(master []byte, account, app, incarnation string) []byte // 32 bytes
+ func Suffix(account, app, incarnation, kid string) string // "acct:app:inc:kid"
+ func Credential(key []byte, username string) string // base64 std
+ type Username struct{ Expiry time.Time; Account, App, Incarnation, Kid string }
+ func ParseUsername(s string) (Username, error)
+ ```
+
+- [ ] **Step 1: Write the vector test**
+
+```go
+package relay
+
+import (
+ "crypto/sha256"
+ "encoding/base64"
+ "testing"
+)
+
+func TestVector(t *testing.T) {
+ m := sha256.Sum256([]byte("carlos-relay-test-master"))
+ key := DeriveKey(m[:], "abc", "meet", "0123456789abcdef")
+ if got := base64.StdEncoding.EncodeToString(key); got != "uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=" {
+ t.Fatalf("key = %s", got)
+ }
+ u := "1900000000:" + Suffix("abc", "meet", "0123456789abcdef", "k1")
+ if u != "1900000000:abc:meet:0123456789abcdef:k1" {
+ t.Fatalf("username = %s", u)
+ }
+ if got := Credential(key, u); got != "jRU479Y3GO45nrRHqxBf3BMeBHE6Fip1cwAccNv28MQ=" {
+ t.Fatalf("credential = %s", got)
+ }
+}
+```
+
+- [ ] **Step 2: Write the parse tests**
+
+```go
+func TestParseUsername(t *testing.T) {
+ u, err := ParseUsername("1900000000:abc:meet:0123456789abcdef:k1")
+ if err != nil || u.Account != "abc" || u.App != "meet" || u.Incarnation != "0123456789abcdef" || u.Kid != "k1" || u.Expiry.Unix() != 1900000000 {
+ t.Fatalf("parse = %+v %v", u, err)
+ }
+ for _, bad := range []string{"", "1:2:3:4", "1:2:3:4:5:6", "x:abc:meet:0123456789abcdef:k1",
+ "1900000000:ABC:meet:0123456789abcdef:k1", "1900000000:abc:9meet:0123456789abcdef:k1",
+ "1900000000:abc:meet:short:k1", "1900000000:abc:meet:0123456789abcdef:K1"} {
+ if _, err := ParseUsername(bad); err == nil {
+ t.Errorf("ParseUsername(%q) accepted", bad)
+ }
+ }
+}
+```
+
+- [ ] **Step 3: Run tests to verify they fail**
+
+Run: `go test ./internal/relay/ -run 'TestVector|TestParseUsername' -v`
+Expected: FAIL, package does not exist.
+
+- [ ] **Step 4: Implement**
+
+```go
+// Package relay is the platform STUN/TURN relay: the app-facing credential
+// contract (spec §2), key derivation (§4), and the server (§3).
+package relay
+
+import (
+ "crypto/hmac"
+ "crypto/sha256"
+ "encoding/base64"
+ "fmt"
+ "regexp"
+ "strconv"
+ "strings"
+ "time"
+
+ "golang.org/x/crypto/hkdf"
+ "io"
+)
+
+const (
+ MaxCredentialLife = 4 * time.Hour
+ hkdfSalt = "carlos-relay-v1"
+)
+
+var (
+ accountRe = regexp.MustCompile(`^[a-z0-9]+$`)
+ appRe = regexp.MustCompile(`^[a-z][a-z0-9-]{0,31}$`)
+ incarnationRe = regexp.MustCompile(`^[0-9a-f]{16}$`)
+ kidRe = regexp.MustCompile(`^[a-z0-9]{1,8}$`)
+)
+
+// DeriveKey is HKDF-SHA256(master, salt "carlos-relay-v1",
+// info account||0x00||app||0x00||incarnation, 32).
+func DeriveKey(master []byte, account, app, incarnation string) []byte {
+ info := []byte(account + "\x00" + app + "\x00" + incarnation)
+ r := hkdf.New(sha256.New, master, []byte(hkdfSalt), info)
+ key := make([]byte, 32)
+ if _, err := io.ReadFull(r, key); err != nil {
+ panic("relay: hkdf: " + err.Error()) // cannot fail for 32 bytes
+ }
+ return key
+}
+
+func Suffix(account, app, incarnation, kid string) string {
+ return account + ":" + app + ":" + incarnation + ":" + kid
+}
+
+// Credential is base64(HMAC-SHA256(key, username)), the TURN password.
+func Credential(key []byte, username string) string {
+ h := hmac.New(sha256.New, key)
+ h.Write([]byte(username))
+ return base64.StdEncoding.EncodeToString(h.Sum(nil))
+}
+
+type Username struct {
+ Expiry time.Time
+ Account, App, Incarnation, Kid string
+}
+
+// ParseUsername splits "<expiry>:<account>:<app>:<incarnation>:<kid>" and
+// validates each field's alphabet. It does not check expiry against now;
+// the auth handler does that with its own clock.
+func ParseUsername(s string) (Username, error) {
+ f := strings.Split(s, ":")
+ if len(f) != 5 {
+ return Username{}, fmt.Errorf("relay: username has %d fields, want 5", len(f))
+ }
+ exp, err := strconv.ParseInt(f[0], 10, 64)
+ if err != nil || exp < 0 {
+ return Username{}, fmt.Errorf("relay: bad expiry %q", f[0])
+ }
+ switch {
+ case !accountRe.MatchString(f[1]):
+ return Username{}, fmt.Errorf("relay: bad account %q", f[1])
+ case !appRe.MatchString(f[2]):
+ return Username{}, fmt.Errorf("relay: bad app %q", f[2])
+ case !incarnationRe.MatchString(f[3]):
+ return Username{}, fmt.Errorf("relay: bad incarnation %q", f[3])
+ case !kidRe.MatchString(f[4]):
+ return Username{}, fmt.Errorf("relay: bad kid %q", f[4])
+ }
+ return Username{Expiry: time.Unix(exp, 0).UTC(), Account: f[1], App: f[2], Incarnation: f[3], Kid: f[4]}, nil
+}
+```
+
+`golang.org/x/crypto` is already in `go.mod`; `hkdf` is a subpackage of it.
+
+- [ ] **Step 5: Run tests to verify they pass**
+
+Run: `go test ./internal/relay/ -v`
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add internal/relay/
+git commit -m "relay: key derivation and the credential contract (spec §2, §4)"
+```
+
+---
+
+### Task 2: The masters file
+
+**Files:**
+- Create: `internal/relay/masters.go`
+- Test: `internal/relay/masters_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ type Master struct{ Kid string; Key []byte; RetireAt time.Time } // zero RetireAt = never
+ type Masters struct{ /* unexported */ }
+ func ParseMasters(r io.Reader) (*Masters, error)
+ func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) // false if absent or retired
+ func (m *Masters) Kids() []string // loaded kids, sorted
+ func WatchMasters(ctx context.Context, path string, onChange func(*Masters, error))
+ ```
+- File format, one master per line: `<kid> <base64 key> <retire_at RFC3339 or "-">`.
+
+- [ ] **Step 1: Write the tests**
+
+```go
+func TestParseMasters(t *testing.T) {
+ in := "k1 " + b64(bytes.Repeat([]byte{1}, 32)) + " -\n" +
+ "k0 " + b64(bytes.Repeat([]byte{2}, 32)) + " 2026-09-08T00:00:00Z\n" +
+ "# comment\n\n"
+ m, err := ParseMasters(strings.NewReader(in))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := m.Kids(); !reflect.DeepEqual(got, []string{"k0", "k1"}) {
+ t.Fatalf("kids = %v", got)
+ }
+ before := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC)
+ after := time.Date(2026, 9, 9, 0, 0, 0, 0, time.UTC)
+ if _, ok := m.Lookup("k0", before); !ok {
+ t.Fatal("k0 should be live before retire_at")
+ }
+ if _, ok := m.Lookup("k0", after); ok {
+ t.Fatal("k0 should be retired after retire_at")
+ }
+ if _, ok := m.Lookup("k1", after); !ok {
+ t.Fatal("k1 never retires")
+ }
+ if _, ok := m.Lookup("kx", before); ok {
+ t.Fatal("unknown kid")
+ }
+}
+
+func TestParseMastersRejectsBadLines(t *testing.T) {
+ 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"} {
+ if _, err := ParseMasters(strings.NewReader(bad)); err == nil {
+ t.Errorf("accepted %q", bad)
+ }
+ }
+}
+
+func b64(b []byte) string { return base64.StdEncoding.EncodeToString(b) }
+```
+
+- [ ] **Step 2: Run to verify failure**
+
+Run: `go test ./internal/relay/ -run TestParseMasters -v`
+Expected: FAIL, undefined.
+
+- [ ] **Step 3: Implement**
+
+```go
+package relay
+
+import (
+ "bufio"
+ "context"
+ "encoding/base64"
+ "fmt"
+ "io"
+ "os"
+ "sort"
+ "strings"
+ "time"
+)
+
+type Master struct {
+ Kid string
+ Key []byte
+ RetireAt time.Time // zero means never
+}
+
+type Masters struct{ byKid map[string]Master }
+
+// ParseMasters reads "<kid> <base64 key> <retire_at|->" lines. Blank lines
+// and "#" comments are skipped. Keys must decode to exactly 32 bytes.
+func ParseMasters(r io.Reader) (*Masters, error) {
+ m := &Masters{byKid: map[string]Master{}}
+ sc := bufio.NewScanner(r)
+ line := 0
+ for sc.Scan() {
+ line++
+ s := strings.TrimSpace(sc.Text())
+ if s == "" || strings.HasPrefix(s, "#") {
+ continue
+ }
+ f := strings.Fields(s)
+ if len(f) != 3 {
+ return nil, fmt.Errorf("relay masters line %d: want 3 fields, got %d", line, len(f))
+ }
+ if !kidRe.MatchString(f[0]) {
+ return nil, fmt.Errorf("relay masters line %d: bad kid %q", line, f[0])
+ }
+ key, err := base64.StdEncoding.DecodeString(f[1])
+ if err != nil || len(key) != 32 {
+ return nil, fmt.Errorf("relay masters line %d: key must be base64 of 32 bytes", line)
+ }
+ var retire time.Time
+ if f[2] != "-" {
+ retire, err = time.Parse(time.RFC3339, f[2])
+ if err != nil {
+ return nil, fmt.Errorf("relay masters line %d: bad retire_at %q", line, f[2])
+ }
+ }
+ m.byKid[f[0]] = Master{Kid: f[0], Key: key, RetireAt: retire}
+ }
+ return m, sc.Err()
+}
+
+func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) {
+ mk, ok := m.byKid[kid]
+ if !ok || (!mk.RetireAt.IsZero() && !now.Before(mk.RetireAt)) {
+ return Master{}, false
+ }
+ return mk, true
+}
+
+func (m *Masters) Kids() []string {
+ out := make([]string, 0, len(m.byKid))
+ for k := range m.byKid {
+ out = append(out, k)
+ }
+ sort.Strings(out)
+ return out
+}
+
+// WatchMasters re-reads path whenever its mtime changes, polling every
+// five seconds (fsnotify would be a dependency for one file). onChange is
+// called with the parse result or error each time the content changes,
+// and once at start.
+func WatchMasters(ctx context.Context, path string, onChange func(*Masters, error)) {
+ var last time.Time
+ tick := time.NewTicker(5 * time.Second)
+ defer tick.Stop()
+ load := func() {
+ st, err := os.Stat(path)
+ if err != nil {
+ onChange(nil, err)
+ return
+ }
+ if st.ModTime().Equal(last) {
+ return
+ }
+ last = st.ModTime()
+ f, err := os.Open(path)
+ if err != nil {
+ onChange(nil, err)
+ return
+ }
+ defer f.Close()
+ onChange(ParseMasters(f))
+ }
+ load()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-tick.C:
+ load()
+ }
+ }
+}
+```
+
+- [ ] **Step 4: Run tests, then commit**
+
+Run: `go test ./internal/relay/ -v`
+Expected: PASS.
+
+```bash
+git add internal/relay/masters.go internal/relay/masters_test.go
+git commit -m "relay: masters file with retire_at and hot reload (spec §4)"
+```
+
+---
+
+### Task 3: Authority record, lease clock, heartbeat
+
+**Files:**
+- Create: `internal/relay/authority.go`
+- Test: `internal/relay/authority_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ const AuthorityKey = "control/relay.json"
+ const LeaseMaxAge = 15 * time.Minute
+ type AuthorityApp struct{ Account, App, Incarnation string; Regions []string }
+ type Authority struct{ Apps []AuthorityApp; WrittenAt time.Time }
+ func (a *Authority) Lookup(account, app string) (AuthorityApp, bool)
+ func ReadAuthority(ctx, s release.Store) (*Authority, string /*etag*/, error) // os.ErrNotExist → nil,"",nil
+ type Lease struct{ /* unexported */ }
+ func (l *Lease) Renew(now time.Time) // on a successful read
+ func (l *Lease) Live(now time.Time) bool // now - last <= LeaseMaxAge
+ func HeartbeatKey(box string) string // "control/relay-ack/<box>.json"
+ type Heartbeat struct{ Box, Region string; WrittenAt time.Time /* of the record held */; ReadAt time.Time; Listeners []string; Kids []string; MastersError string }
+ func WriteHeartbeat(ctx, s release.Store, hb Heartbeat) error // unconditional Put
+ ```
+
+- [ ] **Step 1: Tests**
+
+```go
+func TestAuthorityRoundTrip(t *testing.T) {
+ s := release.NewMemStore() // see internal/release for the in-memory Store used by console tests
+ ctx := context.Background()
+ a, _, err := ReadAuthority(ctx, s)
+ if err != nil || a != nil {
+ t.Fatalf("absent = %v %v, want nil,nil", a, err)
+ }
+ 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()})
+ if _, err := s.Put(ctx, AuthorityKey, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ a, etag, err := ReadAuthority(ctx, s)
+ if err != nil || etag == "" {
+ t.Fatal(err)
+ }
+ if got, ok := a.Lookup("abc", "meet"); !ok || got.Incarnation != "0123456789abcdef" {
+ t.Fatalf("lookup = %+v %v", got, ok)
+ }
+ if _, ok := a.Lookup("abc", "other"); ok {
+ t.Fatal("unknown app found")
+ }
+}
+
+func TestLease(t *testing.T) {
+ var l Lease
+ t0 := time.Unix(1_800_000_000, 0)
+ if l.Live(t0) {
+ t.Fatal("never renewed should not be live")
+ }
+ l.Renew(t0)
+ if !l.Live(t0.Add(14 * time.Minute)) {
+ t.Fatal("14m should be live")
+ }
+ if l.Live(t0.Add(16 * time.Minute)) {
+ t.Fatal("16m should not be live")
+ }
+}
+```
+
+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.
+
+- [ ] **Step 2: Run to verify failure, then implement**
+
+```go
+package relay
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ AuthorityKey = "control/relay.json"
+ LeaseMaxAge = 15 * time.Minute
+)
+
+type AuthorityApp struct {
+ Account string `json:"account"`
+ App string `json:"app"`
+ Incarnation string `json:"incarnation"`
+ Regions []string `json:"regions"`
+}
+
+type Authority struct {
+ Apps []AuthorityApp `json:"apps"`
+ WrittenAt time.Time `json:"written_at"`
+}
+
+func (a *Authority) Lookup(account, app string) (AuthorityApp, bool) {
+ if a == nil {
+ return AuthorityApp{}, false
+ }
+ for _, x := range a.Apps {
+ if x.Account == account && x.App == app {
+ return x, true
+ }
+ }
+ return AuthorityApp{}, false
+}
+
+func ReadAuthority(ctx context.Context, s release.Store) (*Authority, string, error) {
+ body, etag, err := s.Get(ctx, AuthorityKey)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, "", nil
+ }
+ if err != nil {
+ return nil, "", err
+ }
+ var a Authority
+ if err := json.Unmarshal(body, &a); err != nil {
+ return nil, etag, err
+ }
+ return &a, etag, nil
+}
+
+// Lease is the relay's serving lease: live while the last SUCCESSFUL
+// authority read is younger than LeaseMaxAge. A failed read renews nothing.
+type Lease struct{ last time.Time }
+
+func (l *Lease) Renew(now time.Time) { l.last = now }
+func (l *Lease) Live(now time.Time) bool { return !l.last.IsZero() && now.Sub(l.last) <= LeaseMaxAge }
+
+func HeartbeatKey(box string) string { return "control/relay-ack/" + box + ".json" }
+
+type Heartbeat struct {
+ Box string `json:"box"`
+ Region string `json:"region"`
+ WrittenAt time.Time `json:"written_at"` // of the authority record held; zero if none
+ ReadAt time.Time `json:"read_at"`
+ Listeners []string `json:"listeners"`
+ Kids []string `json:"kids"`
+ MastersError string `json:"masters_error,omitempty"`
+}
+
+func WriteHeartbeat(ctx context.Context, s release.Store, hb Heartbeat) error {
+ body, err := json.Marshal(hb)
+ if err != nil {
+ return err
+ }
+ _, err = s.Put(ctx, HeartbeatKey(hb.Box), body, "")
+ return err
+}
+```
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+git add internal/relay/authority.go internal/relay/authority_test.go
+git commit -m "relay: authority record, serving lease, heartbeat (spec §3, §5)"
+```
+
+---
+
+### Task 4: Limits
+
+**Files:**
+- Create: `internal/relay/limits.go`
+- Test: `internal/relay/limits_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ type Limits struct {
+ AppLive, AppPerMinute int; AppBytesPerHour int64
+ AccountLive, AccountPerMinute int; AccountBytesPerHour int64
+ BoxLive, BoxPerMinute int
+ AllocBitsPerSec, BoxBitsPerSec int64
+ }
+ func DefaultLimits() Limits // 64,30,20<<30; 256,60,60<<30; 2048,500; 8_000_000, 200_000_000
+ func LimitsFromEnv(getenv func(string) string) Limits // CARLOS_RELAY_LIMIT_* overrides, names below
+ type Identity struct{ Account, App string }
+ type Admission struct{ /* unexported; one mutex */ }
+ func NewAdmission(l Limits, now func() time.Time) *Admission
+ func (a *Admission) Admit(id Identity) (ok bool, reason string) // counts an attempt even on refusal
+ func (a *Admission) Release(id Identity)
+ func (a *Admission) AddBytes(id Identity, n int64) (over bool) // rolling hour; over=true when a budget is crossed
+ func (a *Admission) Seed(id Identity, at time.Time, n int64) // rebuild rolling window on start
+ func (a *Admission) OverBudget(id Identity, now time.Time) bool
+ type Bucket struct{ /* token bucket */ }
+ func NewBucket(bitsPerSec int64, now func() time.Time) *Bucket
+ func (b *Bucket) Allow(nbytes int) bool
+ ```
+- 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`.
+
+- [ ] **Step 1: Tests**
+
+```go
+func TestAdmissionCaps(t *testing.T) {
+ now := time.Unix(1_800_000_000, 0)
+ clock := func() time.Time { return now }
+ l := DefaultLimits()
+ l.AppLive, l.AppPerMinute, l.AccountLive, l.BoxLive = 2, 100, 3, 100
+ a := NewAdmission(l, clock)
+ meet := Identity{"abc", "meet"}
+ docs := Identity{"abc", "docs"}
+ for i := 0; i < 2; i++ {
+ if ok, r := a.Admit(meet); !ok {
+ t.Fatalf("admit %d refused: %s", i, r)
+ }
+ }
+ if ok, r := a.Admit(meet); ok || r != "app live" {
+ t.Fatalf("third meet = %v %q", ok, r)
+ }
+ if ok, _ := a.Admit(docs); !ok {
+ t.Fatal("docs first should pass")
+ }
+ if ok, r := a.Admit(docs); ok || r != "account live" {
+ t.Fatalf("account cap = %v %q", ok, r)
+ }
+ a.Release(meet)
+ if ok, _ := a.Admit(docs); !ok {
+ t.Fatal("after release, account has room")
+ }
+}
+
+func TestAdmissionRate(t *testing.T) {
+ now := time.Unix(1_800_000_000, 0)
+ clock := func() time.Time { return now }
+ l := DefaultLimits()
+ l.AppPerMinute = 2
+ a := NewAdmission(l, clock)
+ id := Identity{"abc", "meet"}
+ a.Admit(id); a.Release(id)
+ a.Admit(id); a.Release(id)
+ if ok, r := a.Admit(id); ok || r != "app rate" {
+ t.Fatalf("rate = %v %q", ok, r)
+ }
+ now = now.Add(61 * time.Second)
+ if ok, _ := a.Admit(id); !ok {
+ t.Fatal("rate window should have rolled")
+ }
+}
+
+func TestAdmissionBudgetRollsAndSeeds(t *testing.T) {
+ now := time.Unix(1_800_000_000, 0)
+ clock := func() time.Time { return now }
+ l := DefaultLimits()
+ l.AppBytesPerHour = 100
+ a := NewAdmission(l, clock)
+ id := Identity{"abc", "meet"}
+ if a.AddBytes(id, 60) {
+ t.Fatal("60 under 100")
+ }
+ if !a.AddBytes(id, 50) {
+ t.Fatal("110 over 100")
+ }
+ now = now.Add(61 * time.Minute)
+ if a.OverBudget(id, now) {
+ t.Fatal("window rolled")
+ }
+ b := NewAdmission(l, clock)
+ b.Seed(id, now.Add(-30*time.Minute), 90)
+ if !b.AddBytes(id, 20) {
+ t.Fatal("seeded 90 + 20 over 100")
+ }
+}
+
+func TestBucket(t *testing.T) {
+ now := time.Unix(1_800_000_000, 0)
+ clock := func() time.Time { return now }
+ b := NewBucket(8000, clock) // 1000 bytes/s
+ if !b.Allow(1000) {
+ t.Fatal("first second's worth")
+ }
+ if b.Allow(1) {
+ t.Fatal("empty")
+ }
+ now = now.Add(500 * time.Millisecond)
+ if !b.Allow(500) || b.Allow(1) {
+ t.Fatal("refilled 500")
+ }
+}
+```
+
+- [ ] **Step 2: Implement**
+
+```go
+package relay
+
+import (
+ "strconv"
+ "sync"
+ "time"
+)
+
+type Limits struct {
+ AppLive, AppPerMinute int
+ AppBytesPerHour int64
+ AccountLive, AccountPerMinute int
+ AccountBytesPerHour int64
+ BoxLive, BoxPerMinute int
+ AllocBitsPerSec, BoxBitsPerSec int64
+}
+
+func DefaultLimits() Limits {
+ return Limits{
+ AppLive: 64, AppPerMinute: 30, AppBytesPerHour: 20 << 30,
+ AccountLive: 256, AccountPerMinute: 60, AccountBytesPerHour: 60 << 30,
+ BoxLive: 2048, BoxPerMinute: 500,
+ AllocBitsPerSec: 8_000_000, BoxBitsPerSec: 200_000_000,
+ }
+}
+
+func LimitsFromEnv(getenv func(string) string) Limits {
+ l := DefaultLimits()
+ i := func(name string, dst *int) {
+ if v, err := strconv.Atoi(getenv("CARLOS_RELAY_LIMIT_" + name)); err == nil && v > 0 {
+ *dst = v
+ }
+ }
+ i64 := func(name string, dst *int64) {
+ if v, err := strconv.ParseInt(getenv("CARLOS_RELAY_LIMIT_"+name), 10, 64); err == nil && v > 0 {
+ *dst = v
+ }
+ }
+ i("APP_LIVE", &l.AppLive); i("APP_PER_MINUTE", &l.AppPerMinute); i64("APP_BYTES_PER_HOUR", &l.AppBytesPerHour)
+ i("ACCOUNT_LIVE", &l.AccountLive); i("ACCOUNT_PER_MINUTE", &l.AccountPerMinute); i64("ACCOUNT_BYTES_PER_HOUR", &l.AccountBytesPerHour)
+ i("BOX_LIVE", &l.BoxLive); i("BOX_PER_MINUTE", &l.BoxPerMinute)
+ i64("ALLOC_BPS", &l.AllocBitsPerSec); i64("BOX_BPS", &l.BoxBitsPerSec)
+ return l
+}
+
+type Identity struct{ Account, App string }
+
+type window struct { // sliding counters bucketed per minute, 60 buckets for an hour
+ at [60]time.Time
+ n [60]int64
+}
+
+func (w *window) add(now time.Time, n int64) {
+ m := now.Truncate(time.Minute)
+ i := int(m.Unix()/60) % 60
+ if !w.at[i].Equal(m) {
+ w.at[i], w.n[i] = m, 0
+ }
+ w.n[i] += n
+}
+
+func (w *window) sum(now time.Time, span time.Duration) int64 {
+ var s int64
+ cut := now.Add(-span)
+ for i := range w.at {
+ if !w.at[i].IsZero() && w.at[i].After(cut) {
+ s += w.n[i]
+ }
+ }
+ return s
+}
+
+type counters struct {
+ live int
+ starts window // attempts per minute
+ bytes window // bytes per rolling hour
+}
+
+type Admission struct {
+ mu sync.Mutex
+ l Limits
+ now func() time.Time
+ app map[Identity]*counters
+ acct map[string]*counters
+ box counters
+}
+
+func NewAdmission(l Limits, now func() time.Time) *Admission {
+ return &Admission{l: l, now: now, app: map[Identity]*counters{}, acct: map[string]*counters{}}
+}
+
+func (a *Admission) get(id Identity) (*counters, *counters) {
+ c, ok := a.app[id]
+ if !ok {
+ c = &counters{}
+ a.app[id] = c
+ }
+ ac, ok := a.acct[id.Account]
+ if !ok {
+ ac = &counters{}
+ a.acct[id.Account] = ac
+ }
+ return c, ac
+}
+
+// Admit counts an attempt for the app, the account and the box whether or
+// not it is admitted, then checks live caps, rates and budgets in that order.
+func (a *Admission) Admit(id Identity) (bool, string) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ now := a.now()
+ c, ac := a.get(id)
+ c.starts.add(now, 1); ac.starts.add(now, 1); a.box.starts.add(now, 1)
+ switch {
+ case c.live >= a.l.AppLive:
+ return false, "app live"
+ case ac.live >= a.l.AccountLive:
+ return false, "account live"
+ case a.box.live >= a.l.BoxLive:
+ return false, "box live"
+ case c.starts.sum(now, time.Minute) > int64(a.l.AppPerMinute):
+ return false, "app rate"
+ case ac.starts.sum(now, time.Minute) > int64(a.l.AccountPerMinute):
+ return false, "account rate"
+ case a.box.starts.sum(now, time.Minute) > int64(a.l.BoxPerMinute):
+ return false, "box rate"
+ case c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour:
+ return false, "app budget"
+ case ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour:
+ return false, "account budget"
+ }
+ c.live++; ac.live++; a.box.live++
+ return true, ""
+}
+
+func (a *Admission) Release(id Identity) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ c, ac := a.get(id)
+ if c.live > 0 { c.live-- }
+ if ac.live > 0 { ac.live-- }
+ if a.box.live > 0 { a.box.live-- }
+}
+
+// AddBytes records forwarded bytes and reports whether the app or account
+// rolling-hour budget is now exceeded.
+func (a *Admission) AddBytes(id Identity, n int64) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ now := a.now()
+ c, ac := a.get(id)
+ c.bytes.add(now, n); ac.bytes.add(now, n)
+ return c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour || ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour
+}
+
+func (a *Admission) Seed(id Identity, at time.Time, n int64) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ c, ac := a.get(id)
+ c.bytes.add(at, n); ac.bytes.add(at, n)
+}
+
+func (a *Admission) OverBudget(id Identity, now time.Time) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ c, ac := a.get(id)
+ return c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour || ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour
+}
+
+// Bucket is a token bucket in bytes with a one-second burst.
+type Bucket struct {
+ mu sync.Mutex
+ rate float64 // bytes per second
+ tokens float64
+ last time.Time
+ now func() time.Time
+}
+
+func NewBucket(bitsPerSec int64, now func() time.Time) *Bucket {
+ r := float64(bitsPerSec) / 8
+ return &Bucket{rate: r, tokens: r, last: now(), now: now}
+}
+
+func (b *Bucket) Allow(nbytes int) bool {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ t := b.now()
+ b.tokens += t.Sub(b.last).Seconds() * b.rate
+ if b.tokens > b.rate {
+ b.tokens = b.rate
+ }
+ b.last = t
+ if float64(nbytes) > b.tokens {
+ return false
+ }
+ b.tokens -= float64(nbytes)
+ return true
+}
+```
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+git add internal/relay/limits.go internal/relay/limits_test.go
+git commit -m "relay: admission, rolling budgets, token buckets (spec §3 limits)"
+```
+
+---
+
+### Task 5: Wrapped sockets: counting, buckets, EIP port gate, close
+
+**Files:**
+- Create: `internal/relay/conn.go`
+- Test: `internal/relay/conn_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ // Binding is what a relay socket belongs to.
+ type Binding struct{ ID Identity; Expiry time.Time; Kid string; Src string /* typed source key */ }
+ // relayConn wraps an allocation's relay PacketConn.
+ type relayConn struct{ /* unexported */ }
+ 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
+ // WriteTo: client→peer forwarding. Drops (returns n,nil) when the bucket refuses or when dst is the EIP on a port not in ports.
+ // ReadFrom: peer→relay; counted by the server only after pion forwards (see Task 6), so this wrapper counts nothing on read.
+ // Close: idempotent; runs onClose once.
+ type portSet struct{ /* unexported; mutex + map[int]bool */ }
+ func (p *portSet) add(port int); func (p *portSet) remove(port int); func (p *portSet) has(port int) bool
+ func typedSrc(a net.Addr) string // "udp:" or "tcp:" + a.String(); the owner-binding key
+ ```
+
+- [ ] **Step 1: Tests**
+
+```go
+func TestRelayConnGatesEIPPorts(t *testing.T) {
+ inner := &fakePacketConn{}
+ ports := &portSet{m: map[int]bool{50000: true}}
+ eip := net.ParseIP("203.0.113.5")
+ now := time.Unix(1_800_000_000, 0)
+ clock := func() time.Time { return now }
+ var in int
+ c := newRelayConn(inner, &Binding{ID: Identity{"abc", "meet"}}, eip, ports, NewBucket(1<<40, clock), NewBucket(1<<40, clock),
+ func(_ Identity, n int) { in += n }, nil, func(*relayConn) {})
+ ok := &net.UDPAddr{IP: eip, Port: 50000}
+ bad := &net.UDPAddr{IP: eip, Port: 3478}
+ other := &net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 4444}
+ c.WriteTo([]byte("abc"), ok)
+ c.WriteTo([]byte("abc"), bad)
+ c.WriteTo([]byte("abcd"), other)
+ if len(inner.written) != 2 || in != 7 {
+ t.Fatalf("written=%d in=%d, want 2 packets and 7 bytes", len(inner.written), in)
+ }
+}
+
+func TestRelayConnBucketDrops(t *testing.T) {
+ inner := &fakePacketConn{}
+ now := time.Unix(1_800_000_000, 0)
+ clock := func() time.Time { return now }
+ c := newRelayConn(inner, &Binding{}, nil, &portSet{m: map[int]bool{}}, NewBucket(80, clock), NewBucket(1<<40, clock), func(Identity, int) {}, nil, func(*relayConn) {})
+ dst := &net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 4444}
+ c.WriteTo(make([]byte, 10), dst) // 10 bytes = the whole second
+ c.WriteTo(make([]byte, 1), dst)
+ if len(inner.written) != 1 {
+ t.Fatalf("written = %d, want 1 (second dropped by bucket)", len(inner.written))
+ }
+}
+
+func TestRelayConnCloseOnce(t *testing.T) {
+ n := 0
+ c := newRelayConn(&fakePacketConn{}, &Binding{}, nil, &portSet{m: map[int]bool{}}, nil, nil, nil, nil, func(*relayConn) { n++ })
+ c.Close(); c.Close()
+ if n != 1 {
+ t.Fatalf("onClose ran %d times", n)
+ }
+}
+
+func TestTypedSrc(t *testing.T) {
+ u := &net.UDPAddr{IP: net.ParseIP("192.0.2.1"), Port: 5000}
+ tc := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 5000}
+ if typedSrc(u) == typedSrc(tc) {
+ t.Fatal("udp and tcp with the same address must not collide")
+ }
+}
+
+// fakePacketConn records writes; ReadFrom blocks until Close.
+type fakePacketConn struct {
+ mu sync.Mutex
+ written [][]byte
+ closed chan struct{}
+ once sync.Once
+}
+func (f *fakePacketConn) init() { f.once.Do(func() { f.closed = make(chan struct{}) }) }
+func (f *fakePacketConn) ReadFrom(b []byte) (int, net.Addr, error) { f.init(); <-f.closed; return 0, nil, net.ErrClosed }
+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 }
+func (f *fakePacketConn) Close() error { f.init(); select { case <-f.closed: default: close(f.closed) }; return nil }
+func (f *fakePacketConn) LocalAddr() net.Addr { return &net.UDPAddr{IP: net.IPv4zero, Port: 50000} }
+func (f *fakePacketConn) SetDeadline(time.Time) error { return nil }
+func (f *fakePacketConn) SetReadDeadline(time.Time) error { return nil }
+func (f *fakePacketConn) SetWriteDeadline(time.Time) error { return nil }
+```
+
+- [ ] **Step 2: Implement**
+
+```go
+package relay
+
+import (
+ "net"
+ "sync"
+ "time"
+)
+
+// Binding is the identity an allocation was created under. Src is the
+// typed source key (typedSrc) pion will present on later requests.
+type Binding struct {
+ ID Identity
+ Expiry time.Time
+ Kid string
+ Src string
+}
+
+// typedSrc keys the owner binding on transport AND address, because pion
+// tags every allocation's tuple as UDP even for TCP clients and a UDP and a
+// TCP client can share an address string.
+func typedSrc(a net.Addr) string {
+ switch a.(type) {
+ case *net.TCPAddr:
+ return "tcp:" + a.String()
+ default:
+ return "udp:" + a.String()
+ }
+}
+
+type portSet struct {
+ mu sync.Mutex
+ m map[int]bool
+}
+
+func (p *portSet) add(port int) { p.mu.Lock(); p.m[port] = true; p.mu.Unlock() }
+func (p *portSet) remove(port int) { p.mu.Lock(); delete(p.m, port); p.mu.Unlock() }
+func (p *portSet) has(port int) bool { p.mu.Lock(); defer p.mu.Unlock(); return p.m[port] }
+
+// relayConn is the PacketConn the relay hands pion for one allocation.
+// Closing it is how the relay ends the allocation: pion's read loop deletes
+// the allocation when ReadFrom errors (allocation.go:301 in pion/turn
+// v4.1.4). WriteTo is the client→peer leg and is where forwarded bytes are
+// counted, the per-allocation and box buckets applied, and the EIP port
+// gate enforced.
+type relayConn struct {
+ net.PacketConn
+ b *Binding
+ eip net.IP
+ ports *portSet
+ appBucket *Bucket
+ boxBucket *Bucket
+ onIn func(Identity, int)
+ onRejected func(Identity, int)
+ onClose func(*relayConn)
+ once sync.Once
+}
+
+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 {
+ return &relayConn{PacketConn: inner, b: b, eip: eip, ports: ports, appBucket: appBucket, boxBucket: boxBucket,
+ onIn: onIn, onRejected: onRejected, onClose: onClose}
+}
+
+func (c *relayConn) WriteTo(p []byte, addr net.Addr) (int, error) {
+ if ua, ok := addr.(*net.UDPAddr); ok && c.eip != nil && ua.IP.Equal(c.eip) && !c.ports.has(ua.Port) {
+ if c.onRejected != nil {
+ c.onRejected(c.b.ID, len(p))
+ }
+ return len(p), nil // dropped: only live relay ports are reachable on our own EIP
+ }
+ if (c.appBucket != nil && !c.appBucket.Allow(len(p))) || (c.boxBucket != nil && !c.boxBucket.Allow(len(p))) {
+ return len(p), nil // dropped by a token bucket
+ }
+ n, err := c.PacketConn.WriteTo(p, addr)
+ if n > 0 && c.onIn != nil {
+ c.onIn(c.b.ID, n)
+ }
+ return n, err
+}
+
+func (c *relayConn) Close() error {
+ var err error
+ c.once.Do(func() {
+ err = c.PacketConn.Close()
+ if c.onClose != nil {
+ c.onClose(c)
+ }
+ })
+ return err
+}
+```
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+git add internal/relay/conn.go internal/relay/conn_test.go
+git commit -m "relay: wrapped relay socket with counting, buckets and the EIP port gate (spec §3)"
+```
+
+---
+
+### Task 6: The pion server wrapper (Opus)
+
+**Files:**
+- Modify: `go.mod`, `go.sum` (add `github.com/pion/turn/v4 v4.1.4`)
+- Create: `internal/relay/server.go`
+- Test: `internal/relay/server_test.go` (end to end on loopback)
+
+**Interfaces:**
+- Consumes: Tasks 1 to 5.
+- Produces:
+ ```go
+ type ServerConfig struct {
+ Region string
+ RelayIP net.IP // the EIP advertised to clients
+ UDP, TCP string // listen addresses, e.g. ":3478"; empty disables
+ MinPort, MaxPort uint16 // 49152, 65535
+ DenyCIDRs []*net.IPNet // from CARLOS_RELAY_DENY_CIDRS plus the built-in private ranges
+ Masters func() *Masters
+ Authority func() *Authority
+ Lease func(now time.Time) bool
+ Limits Limits
+ Now func() time.Time
+ Log func(string, ...any)
+ Meter *Meter // Task 7; nil allowed in tests
+ }
+ type Server struct{ /* unexported */ }
+ func NewServer(cfg ServerConfig) (*Server, error)
+ func (s *Server) Close() error
+ func (s *Server) EndAll(reason string) // lease lapse, shedding
+ func (s *Server) EndWhere(func(*Binding) bool, reason string) // disable, incarnation change, region removed, budget, master retired
+ func (s *Server) Live() int
+ ```
+
+- [ ] **Step 1: Add the dependency** (sandbox lifted for this one command)
+
+Run: `GOFLAGS= GOPROXY=https://proxy.golang.org go get github.com/pion/turn/v4@v4.1.4 && go mod tidy`
+Then: `GOFLAGS=-mod=mod GOPROXY=off go build ./...` must still pass; `git diff go.mod` shows only pion modules added.
+
+- [ ] **Step 2: Write the end-to-end test**
+
+```go
+func TestServerEndToEnd(t *testing.T) {
+ master := sha256.Sum256([]byte("carlos-relay-test-master"))
+ masters, _ := ParseMasters(strings.NewReader("k1 " + b64(master[:]) + " -\n"))
+ auth := &Authority{Apps: []AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"test"}}}, WrittenAt: time.Now()}
+ now := time.Now()
+ var mu sync.Mutex
+ clock := func() time.Time { mu.Lock(); defer mu.Unlock(); return now }
+ s, err := NewServer(ServerConfig{
+ Region: "test", RelayIP: net.ParseIP("127.0.0.1"), UDP: "127.0.0.1:0",
+ MinPort: 40000, MaxPort: 40100,
+ Masters: func() *Masters { return masters }, Authority: func() *Authority { return auth },
+ Lease: func(time.Time) bool { return true }, Limits: DefaultLimits(), Now: clock, Log: t.Logf,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer s.Close()
+
+ key := DeriveKey(master[:], "abc", "meet", "0123456789abcdef")
+ exp := now.Add(time.Hour).Unix()
+ username := fmt.Sprintf("%d:%s", exp, Suffix("abc", "meet", "0123456789abcdef", "k1"))
+ client := turnClient(t, s.UDPAddr(), username, Credential(key, username))
+ relayConn, err := client.Allocate()
+ if err != nil {
+ t.Fatalf("allocate: %v", err)
+ }
+ defer relayConn.Close()
+ if s.Live() != 1 {
+ t.Fatalf("live = %d", s.Live())
+ }
+
+ // Wrong app's credential from the same source is refused (owner binding).
+ other := fmt.Sprintf("%d:%s", exp, Suffix("abc", "docs", "0123456789abcdef", "k1"))
+ if _, err := turnClient(t, s.UDPAddr(), other, Credential(DeriveKey(master[:], "abc", "docs", "0123456789abcdef"), other)).Allocate(); err == nil {
+ t.Fatal("unknown app allocated")
+ }
+
+ // Expiry ends the allocation with no client action.
+ mu.Lock(); now = now.Add(2 * time.Hour); mu.Unlock()
+ s.tick() // exported for tests: runs the expiry sweep once
+ waitFor(t, func() bool { return s.Live() == 0 })
+}
+
+func TestServerRefusesPrivatePeer(t *testing.T) {
+ // allocate as above, then client.CreatePermission(&net.UDPAddr{IP: net.ParseIP("10.0.0.1"), Port: 1}) must error.
+}
+
+func TestServerEndWhere(t *testing.T) {
+ // allocate, then s.EndWhere(func(b *Binding) bool { return b.ID.App == "meet" }, "disable"); waitFor Live()==0.
+}
+```
+
+`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.
+
+- [ ] **Step 3: Implement `server.go`**
+
+Design, from spec §3:
+
+- `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}]}`.
+- `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`.
+- `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`.
+- `quota(username, realm string, src net.Addr) bool`: `Lease(now)` false → refuse; `Admit(Identity)` → refuse with reason logged.
+- `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)`).
+- `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.
+- `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`).
+- `EndWhere`/`EndAll`: iterate `s.allocs`, `Close()` matching conns.
+- `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`.
+- 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`.
+- 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`.
+
+- [ ] **Step 4: Run the end-to-end tests, then the cross-compile check**
+
+Run: `go test ./internal/relay/ -run TestServer -v`
+Run: `for t in darwin/arm64 windows/amd64 linux/arm64; do GOOS=${t%/*} GOARCH=${t#*/} CGO_ENABLED=0 go build ./... || echo FAIL $t; done`
+Expected: PASS, no FAIL lines.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add go.mod go.sum internal/relay/server.go internal/relay/server_test.go
+git commit -m "relay: pion server with our own address generator, owner binding, termination and peer policy (spec §3, §4)"
+```
+
+---
+
+### Task 7: Meter
+
+**Files:**
+- Create: `internal/relay/meter.go`
+- Test: `internal/relay/meter_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ func MeterKey(box, hour string) string // "metrics/relay/<box>/<hour>.json"
+ type MeterSample struct{ At time.Time; Apps map[string]MeterApp } // key "<account>/<app>"
+ type MeterApp struct{ In, Out, Rejected int64; Allocs int }
+ type MeterChunk struct{ V int; Box, Hour string; Samples []MeterSample }
+ type Meter struct{ /* unexported */ }
+ func NewMeter(box string, now func() time.Time) *Meter
+ func (m *Meter) Add(id Identity, in, out, rejected int64)
+ func (m *Meter) SetAllocs(id Identity, n int)
+ 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
+ func LoadWindow(ctx, s release.Store, box string, now time.Time) ([]MeterSample, error) // previous and current hour's samples
+ ```
+
+- [ ] **Step 1: Tests**
+
+```go
+func TestMeterFlushAppendsDeltas(t *testing.T) {
+ s := newMemStore()
+ now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
+ m := NewMeter("box1", func() time.Time { return now })
+ id := Identity{"abc", "meet"}
+ m.Add(id, 100, 200, 5)
+ if err := m.Flush(context.Background(), s); err != nil { t.Fatal(err) }
+ m.Add(id, 10, 20, 0)
+ if err := m.Flush(context.Background(), s); err != nil { t.Fatal(err) }
+ body, _, _ := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
+ var c MeterChunk
+ json.Unmarshal(body, &c)
+ if len(c.Samples) != 2 || c.Samples[0].Apps["abc/meet"].In != 100 || c.Samples[1].Apps["abc/meet"].Out != 20 {
+ t.Fatalf("chunk = %+v", c)
+ }
+}
+
+func TestLoadWindowSpansTwoHours(t *testing.T) {
+ s := newMemStore()
+ now := time.Date(2026, 9, 7, 12, 1, 0, 0, time.UTC)
+ m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }) // 11:31
+ m.Add(Identity{"abc", "meet"}, 7, 0, 0)
+ m.Flush(context.Background(), s)
+ samples, err := LoadWindow(context.Background(), s, "box1", now)
+ if err != nil || len(samples) != 1 || samples[0].Apps["abc/meet"].In != 7 {
+ t.Fatalf("window = %+v %v", samples, err)
+ }
+}
+```
+
+- [ ] **Step 2: Implement**
+
+```go
+package relay
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/metrics"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func MeterKey(box, hour string) string { return "metrics/relay/" + box + "/" + hour + ".json" }
+
+type MeterApp struct {
+ In int64 `json:"in"`
+ Out int64 `json:"out"`
+ Rejected int64 `json:"rejected"`
+ Allocs int `json:"allocs"`
+}
+
+type MeterSample struct {
+ At time.Time `json:"at"`
+ Apps map[string]MeterApp `json:"apps"`
+}
+
+type MeterChunk struct {
+ V int `json:"v"`
+ Box string `json:"box"`
+ Hour string `json:"hour"`
+ Samples []MeterSample `json:"samples"`
+}
+
+type Meter struct {
+ mu sync.Mutex
+ box string
+ now func() time.Time
+ cur map[Identity]*MeterApp
+}
+
+func NewMeter(box string, now func() time.Time) *Meter {
+ return &Meter{box: box, now: now, cur: map[Identity]*MeterApp{}}
+}
+
+func (m *Meter) get(id Identity) *MeterApp {
+ a, ok := m.cur[id]
+ if !ok {
+ a = &MeterApp{}
+ m.cur[id] = a
+ }
+ return a
+}
+
+func (m *Meter) Add(id Identity, in, out, rejected int64) {
+ m.mu.Lock()
+ a := m.get(id)
+ a.In += in; a.Out += out; a.Rejected += rejected
+ m.mu.Unlock()
+}
+
+func (m *Meter) SetAllocs(id Identity, n int) {
+ m.mu.Lock()
+ m.get(id).Allocs = n
+ m.mu.Unlock()
+}
+
+// Flush writes one sample of deltas since the previous flush, appended to
+// this box's hourly object by read-append-CAS. The relay is the only
+// writer of these objects, so the CAS only ever races with itself.
+func (m *Meter) Flush(ctx context.Context, s release.Store) error {
+ m.mu.Lock()
+ now := m.now()
+ sample := MeterSample{At: now, Apps: map[string]MeterApp{}}
+ for id, a := range m.cur {
+ sample.Apps[id.Account+"/"+id.App] = *a
+ m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept
+ }
+ m.mu.Unlock()
+ hour := metrics.HourOf(now)
+ key := MeterKey(m.box, hour)
+ var lastErr error
+ for attempt := 0; attempt < 3; attempt++ {
+ body, etag, err := s.Get(ctx, key)
+ var c MeterChunk
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ c = MeterChunk{V: 1, Box: m.box, Hour: hour}
+ etag = ""
+ case err != nil:
+ return err
+ default:
+ if json.Unmarshal(body, &c) != nil {
+ c = MeterChunk{V: 1, Box: m.box, Hour: hour}
+ }
+ }
+ c.Samples = append(c.Samples, sample)
+ out, _ := json.Marshal(c)
+ if _, err = s.Put(ctx, key, out, etag); err == nil {
+ return nil
+ } else if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ } else {
+ lastErr = err
+ }
+ }
+ return lastErr
+}
+
+// LoadWindow returns every sample from the previous and current hour's
+// objects, for rebuilding rolling-hour budgets after a restart.
+func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time) ([]MeterSample, error) {
+ var out []MeterSample
+ for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} {
+ body, _, err := s.Get(ctx, MeterKey(box, h))
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ return nil, err
+ }
+ var c MeterChunk
+ if json.Unmarshal(body, &c) == nil {
+ out = append(out, c.Samples...)
+ }
+ }
+ return out, nil
+}
+```
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+git add internal/relay/meter.go internal/relay/meter_test.go
+git commit -m "relay: per-app meter with its own hourly objects (spec §3 metering)"
+```
+
+---
+
+### Task 8: `Serve`, the unit, bootstrap arming, and the adopt restart
+
+**Files:**
+- Create: `internal/relay/serve.go`
+- Create: `internal/host/units/carlos-relay.service`
+- Modify: `internal/host/bootstrap.go:72` (`enabledUnits`), `:96-110` (`systemctlPlan`)
+- Modify: `internal/host/systemapply.go:33` (unit const), `:382` (restart), `:568` (`restartEdge`)
+- Modify: `internal/host/host_test.go:29` (`expectedUnits`)
+- Modify: `relay.go` (root; created here with `serve` and `keygen`, other verbs in Task 11), `main.go:76` (dispatch)
+- Test: `internal/host/host_test.go`, `internal/relay/serve_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ // internal/relay
+ type Config struct {
+ Box, Region string; MastersPath string; PublicIP net.IP; DenyCIDRs []*net.IPNet
+ Store release.Store; Limits Limits; PressurePath string /* /proc/pressure/memory */
+ }
+ func ConfigFromEnv(ctx context.Context, getenv func(string) string, imdsIP, imdsRegion func() string, store release.Store) (Config, error)
+ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error
+ ```
+- 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_*`.
+
+- [ ] **Step 1: The unit file**
+
+```ini
+# The platform STUN/TURN relay (media roadmap stage 4a). Armed only on a
+# box that carries /etc/carlos/relay-masters; ConditionPathExists keeps a
+# box without a master from ever listening. Shares the box with the edge and
+# must lose first: bounded memory, PSI-driven shedding in the process, and
+# the highest OOM adjustment there is.
+[Unit]
+Description=carlos relay (STUN and TURN)
+After=network-online.target
+Wants=network-online.target
+ConditionPathExists=/etc/carlos/relay-masters
+
+[Service]
+EnvironmentFile=-/etc/carlos/host.env
+ExecStart=/opt/carlos/carlos relay serve
+Restart=always
+RestartSec=1
+
+User=carlos
+Group=carlos
+NoNewPrivileges=yes
+ProtectSystem=strict
+ProtectHome=yes
+ReadWritePaths=/data/relay /run/carlos
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectClock=yes
+ProtectHostname=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectKernelLogs=yes
+ProtectControlGroups=yes
+ProtectProc=invisible
+RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+SystemCallArchitectures=native
+MemoryHigh=160M
+MemoryMax=256M
+OOMScoreAdjust=1000
+TasksMax=512
+Nice=5
+CPUWeight=50
+
+[Install]
+WantedBy=multi-user.target
+```
+
+- [ ] **Step 2: Arm it in bootstrap** (`internal/host/bootstrap.go`)
+
+Change line 72:
+```go
+var enabledUnits = []string{"carlos-edge.service", "carlos-relay.service", "carlos-adopt.timer", "carlos-adopt.path", "carlos-stage.path", "carlos-restart.path"}
+```
+Change `systemctlPlan` so the relay is started, not only enabled (timers and paths already are):
+```go
+for _, u := range units {
+ if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") || u == "carlos-relay.service" {
+ armed = append(armed, u)
+ }
+}
+```
+`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).
+
+- [ ] **Step 3: Restart on platform update** (`internal/host/systemapply.go`)
+
+Add beside line 33: `systemRelayUnit = "carlos-relay"`. Add after `restartEdge`:
+```go
+// restartRelay mirrors restartEdge for the relay; a unit that is inactive
+// because its ConditionPathExists failed returns an error from restart,
+// which is not a failure of the update, so it is logged and swallowed.
+func (a *systemApplier) restartRelay() {
+ a.ops.systemctl("reset-failed", systemRelayUnit)
+ if _, err := a.ops.systemctl("try-restart", systemRelayUnit); err != nil {
+ log.Printf("system: relay try-restart: %v", err)
+ }
+}
+```
+Call `a.restartRelay()` immediately after each `a.restartEdge()` succeeds (lines 382 and the rollback site near 692).
+
+- [ ] **Step 4: Tests** in `internal/host/host_test.go`
+
+Add `"carlos-relay.service"` to `expectedUnits` (keep the slice sorted). Add:
+```go
+func TestBootstrapStartsRelayUnit(t *testing.T) {
+ plan := systemctlPlan(enabledUnits)
+ var started []string
+ for _, step := range plan {
+ if step[0] == "start" {
+ started = step[1:]
+ }
+ }
+ if !slices.Contains(started, "carlos-relay.service") {
+ t.Fatalf("relay not started by plan: %v", plan)
+ }
+ if slices.Contains(started, "carlos-edge.service") {
+ t.Fatalf("edge must stay start-by-hand: %v", plan)
+ }
+}
+```
+
+- [ ] **Step 5: `Serve` and `ConfigFromEnv`**
+
+```go
+// Serve runs the relay until ctx is done: watches the masters file, reads
+// authority every minute (and on control/serial change), heartbeats, flushes
+// the meter, ticks expiry, and sheds under memory pressure.
+func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
+ var (
+ mu sync.Mutex
+ masters *Masters
+ mastersE string
+ auth *Authority
+ lease Lease
+ )
+ now := time.Now
+ meter := NewMeter(cfg.Box, now)
+ if samples, err := LoadWindow(ctx, cfg.Store, cfg.Box, now()); err == nil {
+ // seeded into Admission below after NewServer
+ _ = samples
+ }
+ srv, err := NewServer(ServerConfig{
+ Region: cfg.Region, RelayIP: cfg.PublicIP, UDP: ":3478", TCP: ":3478", MinPort: 49152, MaxPort: 65535,
+ DenyCIDRs: cfg.DenyCIDRs, Limits: cfg.Limits, Now: now, Log: logf, Meter: meter,
+ Masters: func() *Masters { mu.Lock(); defer mu.Unlock(); return masters },
+ Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth },
+ Lease: func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) },
+ })
+ if err != nil {
+ return err
+ }
+ defer srv.Close()
+ // (seed: for each sample in the window, srv.admission.Seed(id, sample.At, app.In+app.Out))
+
+ go WatchMasters(ctx, cfg.MastersPath, func(m *Masters, err error) {
+ mu.Lock()
+ if err != nil { mastersE = err.Error() } else { masters, mastersE = m, "" }
+ mu.Unlock()
+ if err == nil { srv.tick() } // retire_at may have passed
+ })
+
+ readAuth := func() {
+ a, _, err := ReadAuthority(ctx, cfg.Store)
+ t := now()
+ mu.Lock()
+ if err == nil { auth = a; lease.Renew(t) } else { logf("relay: authority read: %v", err) }
+ live := lease.Live(t)
+ hb := Heartbeat{Box: cfg.Box, Region: cfg.Region, ReadAt: t, Listeners: srv.Listeners(), Kids: kidsOf(masters), MastersError: mastersE}
+ if auth != nil { hb.WrittenAt = auth.WrittenAt }
+ mu.Unlock()
+ if !live { srv.EndAll("lease lapsed") }
+ srv.EndWhere(func(b *Binding) bool { // authority changed under it
+ x, ok := auth.Lookup(b.ID.Account, b.ID.App)
+ return !ok || x.Incarnation != b.Incarnation || !slices.Contains(x.Regions, cfg.Region)
+ }, "authority")
+ if err := WriteHeartbeat(ctx, cfg.Store, hb); err != nil { logf("relay: heartbeat: %v", err) }
+ }
+ readAuth()
+ authTick := time.NewTicker(time.Minute)
+ serialTick := time.NewTicker(3 * time.Second)
+ meterTick := time.NewTicker(time.Minute)
+ expiryTick := time.NewTicker(10 * time.Second)
+ psiTick := time.NewTicker(5 * time.Second)
+ defer func() { for _, t := range []*time.Ticker{authTick, serialTick, meterTick, expiryTick, psiTick} { t.Stop() } }()
+ var lastSerial int64 = -1
+ for {
+ select {
+ case <-ctx.Done():
+ return nil
+ case <-authTick.C:
+ readAuth()
+ case <-serialTick.C:
+ if doc, _, err := release.ReadSerial(ctx, cfg.Store); err == nil && doc.Serial != lastSerial {
+ lastSerial = doc.Serial
+ readAuth()
+ }
+ case <-meterTick.C:
+ if err := meter.Flush(ctx, cfg.Store); err != nil { logf("relay: meter flush: %v", err) }
+ case <-expiryTick.C:
+ srv.tick()
+ case <-psiTick.C:
+ if some, err := readPSI(cfg.PressurePath); err == nil {
+ if some > 20 { srv.ShedOldest(1) } // one per tick until under 10; ShedOldest is a Server method closing the oldest binding
+ }
+ }
+ }
+}
+```
+`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.
+
+- [ ] **Step 6: The verb**
+
+Create `relay.go` at the root with `runRelay(args)` dispatching `serve` and `keygen` now (other sub-verbs in Task 11):
+```go
+func runRelayServe(args []string) {
+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+ defer stop()
+ store, err := openDeployStoreForCLI(ctx) // see below
+ if err != nil { log.Fatalf("relay serve: %v", err) }
+ cfg, err := relay.ConfigFromEnv(ctx, os.Getenv, host.ReadPublicIP, host.ReadPlacementRegion, store)
+ if err != nil { log.Fatalf("relay serve: %v", err) }
+ if err := relay.Serve(ctx, cfg, log.Printf); err != nil { log.Fatalf("relay serve: %v", err) }
+}
+
+func runRelayKeygen(_ []string) {
+ key := make([]byte, 32)
+ rand.Read(key)
+ kid := make([]byte, 4)
+ rand.Read(kid)
+ fmt.Printf("%s %s -\n", hex.EncodeToString(kid), base64.StdEncoding.EncodeToString(key))
+}
+```
+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()`.
+
+- [ ] **Step 7: Run all tests, cross-compile, commit**
+
+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`
+
+```bash
+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
+git commit -m "relay: carlos relay serve, the unit, bootstrap arming and adopt restart (spec §3)"
+```
+
+---
+
+### Task 9: The per-app record and the config value
+
+**Files:**
+- Create: `internal/relay/record.go`, `internal/relay/config.go`
+- Test: `internal/relay/record_test.go`, `internal/relay/config_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ func RecordKey(acct, app string) string // "apps/<acct>/<app>/relay.json"
+ type RouteCutover struct{ Host string; Suffixes map[string]string /* region→suffix */; At time.Time; By string }
+ type Record struct {
+ Enabled bool; Incarnation string; Regions []string
+ Op int64; Phase string /* "config"|"aggregate"|"done" */
+ IssuedSuffixes map[string]string /* region→username_suffix last issued */
+ Cutovers []RouteCutover /* member acks for TCP exec routes */
+ EnabledAt, DisabledAt time.Time
+ }
+ func GetRecord(ctx, s release.Store, acct, app string) (*Record, string, error)
+ func PutRecord(ctx, s release.Store, acct, app string, r *Record, etag string) (string, error)
+ func NewIncarnation() string // 16 lowercase hex from crypto/rand
+ type RegionMaster struct{ Region, Kid string; Master []byte }
+ type RegionURLs func(region string) []string // urls for a region, from the platform domain
+ func BuildConfig(acct, app string, r *Record, masters []RegionMaster, urls RegionURLs) (value string, suffixes map[string]string, err error)
+ // 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.
+ func ParseConfig(value string) ([]ConfigEntry, error) // for the config-ack (Task 12) and tests
+ type ConfigEntry struct{ Region string; URLs []string; Key string; UsernameSuffix string }
+ ```
+
+- [ ] **Step 1: Tests**
+
+```go
+func TestBuildConfigMatchesContract(t *testing.T) {
+ m := sha256.Sum256([]byte("carlos-relay-test-master"))
+ r := &Record{Enabled: true, Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}
+ urls := func(region string) []string {
+ h := "relay." + region + ".oncarlos.com:3478"
+ return []string{"stun:" + h, "turn:" + h + "?transport=udp", "turn:" + h + "?transport=tcp"}
+ }
+ val, suffixes, err := BuildConfig("abc", "meet", r, []RegionMaster{{Region: "eu-west-1", Kid: "k1", Master: m[:]}}, urls)
+ if err != nil { t.Fatal(err) }
+ if err := appconfig.ValidVarValue(val); err != nil { t.Fatalf("not env-safe: %v", err) }
+ entries, err := ParseConfig(val)
+ if err != nil || len(entries) != 1 { t.Fatalf("parse = %+v %v", entries, err) }
+ e := entries[0]
+ if e.Key != "uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=" || e.UsernameSuffix != "abc:meet:0123456789abcdef:k1" || suffixes["eu-west-1"] != e.UsernameSuffix {
+ t.Fatalf("entry = %+v", e)
+ }
+ if len(e.URLs) != 3 || e.URLs[0] != "stun:relay.eu-west-1.oncarlos.com:3478" { t.Fatalf("urls = %v", e.URLs) }
+}
+
+func TestBuildConfigRefusesMissingMaster(t *testing.T) {
+ r := &Record{Enabled: true, Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1", "us-east-1"}}
+ if _, _, err := BuildConfig("abc", "meet", r, []RegionMaster{{Region: "eu-west-1", Kid: "k1", Master: make([]byte, 32)}}, func(string) []string { return nil }); err == nil {
+ t.Fatal("missing us-east-1 master accepted")
+ }
+}
+
+func TestNewIncarnation(t *testing.T) {
+ a, b := NewIncarnation(), NewIncarnation()
+ if !incarnationRe.MatchString(a) || a == b { t.Fatalf("%q %q", a, b) }
+}
+
+func TestRecordRoundTrip(t *testing.T) {
+ s := newMemStore()
+ ctx := context.Background()
+ r, etag, err := GetRecord(ctx, s, "abc", "meet")
+ if err != nil || r != nil || etag != "" { t.Fatalf("absent = %v %q %v", r, etag, err) }
+ etag, err = PutRecord(ctx, s, "abc", "meet", &Record{Enabled: true, Incarnation: NewIncarnation(), Op: 1, Phase: "config"}, "")
+ if err != nil { t.Fatal(err) }
+ if _, err := PutRecord(ctx, s, "abc", "meet", &Record{}, "stale-etag"); !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("stale put = %v, want CAS conflict", err)
+ }
+ r, _, _ = GetRecord(ctx, s, "abc", "meet")
+ if !r.Enabled || r.Op != 1 { t.Fatalf("record = %+v", r) }
+}
+```
+
+- [ ] **Step 2: Implement** `record.go` (mirror `internal/objstore/objstore.go:132-170` Get/Put with `os.ErrNotExist` → nil) and `config.go`:
+
+```go
+type ConfigEntry struct {
+ Region string `json:"region"`
+ URLs []string `json:"urls"`
+ Key string `json:"key"`
+ UsernameSuffix string `json:"username_suffix"`
+}
+
+func BuildConfig(acct, app string, r *Record, masters []RegionMaster, urls RegionURLs) (string, map[string]string, error) {
+ byRegion := map[string]RegionMaster{}
+ for _, m := range masters { byRegion[m.Region] = m }
+ entries := make([]ConfigEntry, 0, len(r.Regions))
+ suffixes := map[string]string{}
+ for _, region := range r.Regions {
+ m, ok := byRegion[region]
+ if !ok { return "", nil, fmt.Errorf("relay: no issuer master for region %s", region) }
+ key := DeriveKey(m.Master, acct, app, r.Incarnation)
+ suffix := Suffix(acct, app, r.Incarnation, m.Kid)
+ suffixes[region] = suffix
+ entries = append(entries, ConfigEntry{Region: region, URLs: urls(region), Key: base64.StdEncoding.EncodeToString(key), UsernameSuffix: suffix})
+ }
+ js, err := json.Marshal(entries)
+ if err != nil { return "", nil, err }
+ return base64.StdEncoding.EncodeToString(js), suffixes, nil
+}
+
+func ParseConfig(value string) ([]ConfigEntry, error) {
+ js, err := base64.StdEncoding.DecodeString(value)
+ if err != nil { return nil, fmt.Errorf("relay: CARLOS_RELAY is not base64: %w", err) }
+ var out []ConfigEntry
+ if err := json.Unmarshal(js, &out); err != nil { return nil, fmt.Errorf("relay: CARLOS_RELAY is not a JSON array: %w", err) }
+ return out, nil
+}
+```
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+git add internal/relay/record.go internal/relay/config.go internal/relay/*_test.go
+git commit -m "relay: per-app record and the CARLOS_RELAY value (spec §2, §5)"
+```
+
+---
+
+### Task 10: Console operations: writer lease, aggregate publish, reconcile (Opus)
+
+**Files:**
+- Create: `internal/console/relayops.go`
+- Test: `internal/console/relayops_test.go`
+
+**Interfaces:**
+- Consumes: Task 9; `appconfig.Apply` (`internal/appconfig/generation.go:119`), `appconfig.Seal`, `release.BumpSerial`, `s.cfg.Releases`, `s.cfg.SecretsPub`.
+- Produces (methods on `*Server`, all requiring the writer lease):
+ ```go
+ const relayWriterKey = "control/relay-writer.json"
+ const relayIssuerKey = "control/relay-issuer.json" // {"<region>":"<kid>"}
+ type relayWriter struct{ /* holder id, expiry; renew every 20s; 60s TTL */ }
+ func (s *Server) relayWriterLoop(ctx) // acquires/renews; sets s.relayHeld atomically
+ func (s *Server) relayHold() bool
+ func (s *Server) relayMasters(region string) (kid string, master []byte, err error) // console secret CARLOS_RELAY_MASTERS_<REGION> + issuer pointer
+ func (s *Server) relayEnable(ctx, acct *Account, app string, regions []string, by string) (*relay.Record, error)
+ func (s *Server) relayDisable(ctx, acct *Account, app string, by string) (*relay.Record, error)
+ func (s *Server) relayAdopted(ctx, acct *Account, app, host string, by string) (*relay.Record, error)
+ func (s *Server) relayRotate(ctx, region string) error // gate on heartbeats, move issuer pointer, re-issue every enabled app in region
+ func (s *Server) relayPublishAggregate(ctx) error // etag-first rebuild from a listing, CAS, retry
+ func (s *Server) relayReconcile(ctx) error // every 5 min and at start; converges config and aggregate from every record
+ ```
+- 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>`).
+
+- [ ] **Step 1: Tests** (each with the console fixture; `fx.rel` is the `release.Store`)
+
+```go
+func TestRelayEnableWritesRecordConfigAndAggregate(t *testing.T) {
+ fx, _ := relayFixture(t, "code-relay-enable") // like storeFixture plus RelayRegions=["eu-west-1"], a test master in the fixture's console secrets, SecretsPub set
+ ctx := context.Background()
+ rec, err := fx.srv.relayEnable(ctx, fx.acct(), "titogo", nil, "alice@keymail.dev")
+ if err != nil { t.Fatal(err) }
+ if !rec.Enabled || rec.Phase != "done" || rec.Op != 1 || len(rec.Incarnation) != 16 { t.Fatalf("record = %+v", rec) }
+ m, _, _ := appconfig.Read(ctx, fx.rel, fx.sqid, "titogo") // whatever reads the default bundle manifest
+ if v, ok := m.Vars["CARLOS_RELAY"]; !ok || v.Kind != "secret" { t.Fatalf("CARLOS_RELAY = %+v", v) }
+ a, _, _ := relay.ReadAuthority(ctx, fx.rel)
+ if x, ok := a.Lookup(fx.sqid, "titogo"); !ok || x.Incarnation != rec.Incarnation { t.Fatalf("aggregate = %+v", a) }
+}
+
+func TestRelayRepeatEnableKeepsIncarnationAndWritesNoConfig(t *testing.T) { /* enable twice; second returns same Incarnation, Op unchanged, config generation unchanged */ }
+
+func TestRelayDisableThenEnableIsANewIncarnation(t *testing.T) { /* enable, disable (aggregate lacks app, CARLOS_RELAY unset), enable: Incarnation differs */ }
+
+func TestRelayPublishAggregateIsEtagFirst(t *testing.T) {
+ // Seed two enabled records A and B. Read the aggregate etag. Disable A directly via PutRecord + relayPublishAggregate.
+ // 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.
+}
+
+func TestRelayReconcileFinishesInterruptedEnable(t *testing.T) {
+ // PutRecord{Enabled:true, Op:1, Phase:"config"} with no config and no aggregate; relayReconcile; expect CARLOS_RELAY set, aggregate has app, Phase "done".
+}
+
+func TestRelayReconcileUndoesStaleConfigAfterDisable(t *testing.T) {
+ // Disabled record but CARLOS_RELAY present in config; relayReconcile; expect it unset.
+}
+
+func TestRelayReconcileRepairsDeletionWithoutRecord(t *testing.T) {
+ // Aggregate lists an app with no record; relayReconcile; aggregate no longer lists it.
+}
+
+func TestRelayEnableRefusesRouteWithoutDelivery(t *testing.T) {
+ // fx.addRoute(registry.Route{Host: "x.titogo...", App: "titogo", Addr: "10.0.0.5:8080"}) with no exec_delivery; relayEnable errors naming the host.
+}
+
+func TestRelayRotateRefusesUntilHeartbeatsCarryKid(t *testing.T) {
+ // 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".
+}
+
+func TestRelayWriterLeaseExcludesSecondHolder(t *testing.T) {
+ // Two Server values over one store; first acquires; second's acquire fails until the first's expiry passes on a fake clock.
+}
+```
+
+- [ ] **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`.
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+git add internal/console/relayops.go internal/console/relayops_test.go cmd/console/main.go internal/console/server.go
+git commit -m "console: relay operations under a writer lease, etag-first aggregate, converging reconcile (spec §5)"
+```
+
+---
+
+### Task 11: Console doors and the CLI
+
+**Files:**
+- Create: `internal/console/apirelay.go`
+- 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`)
+- Test: `internal/console/apirelay_test.go`
+
+**Interfaces:**
+- Routes (`s.cliRoute`, `svcrole.Allow(svcrole.Admin)` for all but status which allows read roles too, matching store):
+ ```
+ POST /api/cli/apps/{account}/{app}/relay body {"regions":[...]} → relayEnable
+ GET /api/cli/apps/{account}/{app}/relay → status
+ POST /api/cli/apps/{account}/{app}/relay/disable → relayDisable
+ POST /api/cli/apps/{account}/{app}/relay/adopted body {"host":"..."} → relayAdopted
+ POST /api/cli/relay/rotate body {"region":"..."} → relayRotate (operator: CARLOS_RELAY_OPERATORS, checked after auth, byte-identical 404 on failure)
+ ```
+- Status reply shape:
+ ```go
+ type apiRelayReply struct {
+ Enabled bool; Incarnation string; Regions []string; Op int64; Phase string
+ Issued map[string]string // region → username_suffix
+ Routes []apiRelayRoute // per route: Host, Shape, Delivery, AckedAt, Cutover
+ Regions []apiRelayRegion // Region, HeartbeatAge string, Current bool, LeaseLive bool, Listeners, Kids []string, Stale bool
+ DisablePending []string // regions not yet past the disable
+ RetireNoEarlierThan map[string]string // region → RFC3339, when rotate is in progress
+ }
+ ```
+
+- [ ] **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.
+
+- [ ] **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.
+
+- [ ] **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.
+
+- [ ] **Step 4: Docs**: add to `docs/site/cli.md` under "Your apps", after `### carlos store`:
+
+````markdown
+### carlos relay
+
+A STUN and TURN relay the platform runs for your app's WebRTC calls, so a
+browser can find its public address and, when two networks cannot reach
+each other directly, relay through the deployment instead of a third party.
+
+```sh
+carlos relay enable --app meet
+carlos relay status --app meet
+```
+
+`enable` turns it on for one app and delivers `CARLOS_RELAY` to the app's
+config: a base64-wrapped JSON list of regions, each with its ICE server URLs,
+a key and a username suffix. The app mints a TURN credential per call from
+the key (see the relay design spec, §2). `status` shows what was issued,
+which routes have picked it up, and the health of each region's relay,
+none of which needs box access. `disable` removes it and ends every
+allocation the app holds. `adopted --host <route>` is how a member tells the
+platform that a TCP-address route it restarted by hand now runs the new
+key. `rotate --region <r>` is the operator's verb after staging a new
+master; `keygen` prints one. `serve` is the unit's entry point and not a
+thing to run by hand.
+````
+
+- [ ] **Step 5: Run the docsite gate and all tests, commit**
+
+Run: `go test ./internal/docsite/ ./internal/console/ -count=1 && go build ./...`
+
+```bash
+git add internal/console/apirelay.go internal/console/apirelay_test.go internal/console/authcli.go apimode.go relay.go docs/site/cli.md
+git commit -m "relay: console doors and carlos relay enable|status|disable|adopted|rotate (spec §5)"
+```
+
+---
+
+### Task 12: The per-route config acknowledgement
+
+**Files:**
+- Create: `internal/appconfig/configack.go`
+- Modify: `internal/appconfig/sync.go:331-434` (`syncHost` gains an ack writer callback), `internal/host/agent.go:2937` (pass the store and box)
+- Test: `internal/appconfig/configack_test.go`, `internal/appconfig/sync_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ func ConfigAckKey(box, host string) string // "control/config-ack/<box>/<host>.json"
+ type ConfigAck struct {
+ Box, Host, Shape string // shape: "unit" | "socket-exec" | "tcp-exec-file"
+ At time.Time
+ Restarted bool; RestartError string
+ RelaySuffixes map[string]string // region → username_suffix parsed from CARLOS_RELAY, nil if absent
+ }
+ func WriteConfigAck(ctx, s release.Store, ack ConfigAck) error
+ // Sync gains a parameter: ack func(ConfigAck) (nil = no acks)
+ func Sync(ctx, reg, s, priv, envDir string, restart func(host string) error, ack func(ConfigAck)) error
+ ```
+
+- [ ] **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`.
+
+- [ ] **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) } }`.
+
+- [ ] **Step 3: Run tests, commit**
+
+```bash
+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
+git commit -m "appconfig: per-route config acknowledgement naming rendered relay identities (spec §5)"
+```
+
+---
+
+### Task 13: Infrastructure
+
+**Files:**
+- Modify: `infra/modules/region-host/variables.tf`, `main.tf`, `outputs.tf`
+- Modify (pinfra repo, separate PR there): `flagship/region-ie.tf` and siblings; `host-init.sh` if it stages `host.env`
+
+- [ ] **Step 1: Module**: add
+```hcl
+variable "relay" {
+ 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."
+ type = bool
+ default = false
+}
+```
+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.
+
+- [ ] **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`.
+
+- [ ] **Step 3: Commit** (platform repo part)
+
+```bash
+git add infra/modules/region-host/
+git commit -m "infra: relay variable opens stun/turn ingress and outputs the VPC CIDR (spec §6)"
+```
+
+---
+
+### Task 14: Whole-branch gate and landing
+
+- [ ] **Step 1: Full gate**: `go build ./... && go vet ./... && go test ./... -count=1 && gofmt -l .` (empty output) and the three cross-compiles.
+- [ ] **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.
+- [ ] **Step 3: Opus review** of the whole branch against the spec, with `superpowers:requesting-code-review`.
+- [ ] **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>`.
+- [ ] **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.
+
+---
+
+## Self-review
+
+**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.
+
+**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".
+
+**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`.