amadan

carlos / platform Public

relay: per-app record and the CARLOS_RELAY value (spec §2, §5)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 1987052d5b5e45b2b3e9e6b9342e317a68e997a4 parent 5f646e0
4 files changed, +244 −0
  • internal/relay/config.go +71 −0
  • internal/relay/config_test.go +42 −0
  • internal/relay/record.go +95 −0
  • internal/relay/record_test.go +36 −0
diff --git a/internal/relay/config.go b/internal/relay/config.go
new file mode 100644
index 000000000..b0dbc1b2d
--- /dev/null
+++ b/internal/relay/config.go
@@ -0,0 +1,71 @@
+package relay
+
+import (
+ "encoding/base64"
+ "encoding/json"
+ "fmt"
+)
+
+// RegionMaster is one region's active issuer master (spec §4): the master
+// key bytes and the kid it's issued under.
+type RegionMaster struct {
+ Region string
+ Kid string
+ Master []byte
+}
+
+// RegionURLs returns the STUN/TURN URLs for a region, built from the
+// platform domain.
+type RegionURLs func(region string) []string
+
+// ConfigEntry is one region's worth of the CARLOS_RELAY value (spec §2):
+// the URLs to try, the derived key, and the username suffix a client
+// appends its own expiry to.
+type ConfigEntry struct {
+ Region string `json:"region"`
+ URLs []string `json:"urls"`
+ Key string `json:"key"`
+ UsernameSuffix string `json:"username_suffix"`
+}
+
+// BuildConfig renders CARLOS_RELAY's value for r: base64(std, padded) of
+// the JSON array from spec §2, one entry per region in r.Regions order.
+// masters must cover every region in r.Regions, or BuildConfig refuses
+// rather than issue a config a region can't actually serve.
+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
+}
+
+// ParseConfig decodes a CARLOS_RELAY value back into its entries, for the
+// config-ack (Task 12) and tests.
+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
+}
diff --git a/internal/relay/config_test.go b/internal/relay/config_test.go
new file mode 100644
index 000000000..3e0745d59
--- /dev/null
+++ b/internal/relay/config_test.go
@@ -0,0 +1,42 @@
+package relay
+
+import (
+ "crypto/sha256"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+)
+
+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")
+ }
+}
diff --git a/internal/relay/record.go b/internal/relay/record.go
new file mode 100644
index 000000000..11db01a74
--- /dev/null
+++ b/internal/relay/record.go
@@ -0,0 +1,95 @@
+package relay
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// RecordSchema is the current Record schema version.
+const RecordSchema = 1
+
+// RecordKey is the per-app record's location in the deployment bucket.
+func RecordKey(acct, app string) string {
+ return release.AppPrefix(acct, app) + "relay.json"
+}
+
+// RouteCutover is one member's ack for a TCP exec route (spec §5): the
+// address they're cutting to, the per-region suffixes they've picked up,
+// and who/when.
+type RouteCutover struct {
+ Host string `json:"host"`
+ Suffixes map[string]string `json:"suffixes"` // region -> suffix
+ At time.Time `json:"at"`
+ By string `json:"by"`
+}
+
+// Record is the per-app relay state (spec §2, §5): whether the relay is
+// enabled for the app, its current key incarnation, the regions it's live
+// in, the in-flight rotation op (if any), and the bookkeeping a rotation
+// needs to converge.
+type Record struct {
+ Schema int `json:"schema"`
+ Enabled bool `json:"enabled"`
+ Incarnation string `json:"incarnation"`
+ Regions []string `json:"regions"`
+ Op int64 `json:"op"`
+ Phase string `json:"phase"` // "config"|"aggregate"|"done"
+ IssuedSuffixes map[string]string `json:"issued_suffixes"` // region -> username_suffix last issued
+ Cutovers []RouteCutover `json:"cutovers"` // member acks for TCP exec routes
+ EnabledAt time.Time `json:"enabled_at"`
+ DisabledAt time.Time `json:"disabled_at"`
+}
+
+// GetRecord reads the record. Absent is (nil, "", nil) — same idiom as
+// objstore.Get, because "no relay" is a state, not an error.
+func GetRecord(ctx context.Context, s release.Store, acct, app string) (*Record, string, error) {
+ if !release.ValidAccount(acct) {
+ return nil, "", fmt.Errorf("relay: invalid account %q", acct)
+ }
+ if !release.ValidApp(app) {
+ return nil, "", fmt.Errorf("relay: invalid app %q", app)
+ }
+ raw, etag, err := s.Get(ctx, RecordKey(acct, app))
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, "", nil
+ }
+ return nil, "", err
+ }
+ var rec Record
+ if err := json.Unmarshal(raw, &rec); err != nil {
+ return nil, "", fmt.Errorf("relay: %s: %w", RecordKey(acct, app), err)
+ }
+ return &rec, etag, nil
+}
+
+// PutRecord writes the record with the caller's CAS position; etag "" is
+// create-only. Every writer is a read-modify-write, so a lost race is
+// release.ErrCASConflict for the caller to surface, never to absorb.
+func PutRecord(ctx context.Context, s release.Store, acct, app string, r *Record, etag string) (string, error) {
+ r.Schema = RecordSchema
+ body, err := json.MarshalIndent(r, "", " ")
+ if err != nil {
+ return "", err
+ }
+ return s.Put(ctx, RecordKey(acct, app), append(body, '\n'), etag)
+}
+
+// NewIncarnation is 16 lowercase hex characters from crypto/rand — a fresh
+// key incarnation, distinct from every prior one with overwhelming
+// probability.
+func NewIncarnation() string {
+ var b [8]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ panic("relay: crypto/rand: " + err.Error())
+ }
+ return hex.EncodeToString(b[:])
+}
diff --git a/internal/relay/record_test.go b/internal/relay/record_test.go
new file mode 100644
index 000000000..de6345014
--- /dev/null
+++ b/internal/relay/record_test.go
@@ -0,0 +1,36 @@
+package relay
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+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(t)
+ 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)
+ }
+}