amadan

carlos / platform Public

registry: fleets + fleet_boxes tables — account-owned fleet records, hashed tokens

Paul Campbell a2f5f619328fca9aa7dba713312587212be29b22 parent bef102c
3 files changed, +512 −10
  • internal/registry/fleets.go +274 −0
  • internal/registry/fleets_test.go +205 −0
  • internal/registry/registry.go +33 −10
diff --git a/internal/registry/fleets.go b/internal/registry/fleets.go
new file mode 100644
index 000000000..e4a865baa
--- /dev/null
+++ b/internal/registry/fleets.go
@@ -0,0 +1,274 @@
+// fleets.go: customer fleets (2026-08-05-customer-fleets-design.md) — an
+// account-owned group of remote boxes that dial this host over a channel
+// identified by "<fleet>/<label>". Two tables: fleets (the account-owned
+// record: which bucket it belongs to) and fleet_boxes (the boxes registered
+// to it, each carrying a hashed bearer token). This package never sees a
+// plaintext token — minting happens console-side, in a later task; here it
+// is always TokenHash, the sha256 hex.
+package registry
+
+import (
+ "database/sql"
+ "fmt"
+ "time"
+)
+
+// Fleet is one account-owned collection of remote boxes.
+type Fleet struct {
+ Name string // ValidTunnelName
+ Account string // owning account identity, same domain as Route.Account
+ Bucket string // customer data bucket name
+ CreatedAt int64
+}
+
+// FleetBox is one box registered to a Fleet. (Fleet, Label) is the channel
+// identity a remote-host-agent dials in as ("<fleet>/<label>" — see
+// FleetChannelAuth); TokenHash is the sha256 hex of its bearer token, never
+// the plaintext.
+type FleetBox struct {
+ Fleet string
+ Label string // ValidTunnelName; (Fleet, Label) is the channel identity
+ TokenHash string // sha256 hex of the bearer token; never plaintext
+ Status string // "active" | "detached"
+ AddedAt int64
+}
+
+// migrateFleets creates the fleets and fleet_boxes tables — additive
+// (CREATE TABLE IF NOT EXISTS), called from migrate() beside the routes DDL,
+// so deploying this code over an existing registry is always safe.
+func (r *Registry) migrateFleets() error {
+ if _, err := r.sql.Exec(`
+CREATE TABLE IF NOT EXISTS fleets (
+ name TEXT PRIMARY KEY,
+ account TEXT NOT NULL,
+ bucket TEXT NOT NULL,
+ created_at INTEGER NOT NULL
+);`); err != nil {
+ return err
+ }
+ _, err := r.sql.Exec(`
+CREATE TABLE IF NOT EXISTS fleet_boxes (
+ fleet TEXT NOT NULL,
+ label TEXT NOT NULL,
+ token_hash TEXT NOT NULL,
+ status TEXT NOT NULL,
+ added_at INTEGER NOT NULL,
+ PRIMARY KEY (fleet, label)
+);`)
+ return err
+}
+
+// freshFleetsConn opens a NEW connection over the same file — the List /
+// Fingerprint pattern (registry.go:455): modernc/sqlite's long-lived pool can
+// serve a stale WAL snapshot for another process's commit. Every read below
+// that a later task polls cross-process (FleetChannelAuth, read by the edge
+// after Fingerprint moves) or that a short-lived CLI process may run moments
+// after a different process wrote (ListFleets, ListFleetBoxes) goes through
+// this, not r.sql.
+func (r *Registry) freshFleetsConn() (*sql.DB, error) {
+ db, err := sql.Open("sqlite", r.path)
+ if err != nil {
+ return nil, err
+ }
+ if _, err := db.Exec("PRAGMA busy_timeout=5000"); err != nil {
+ db.Close()
+ return nil, err
+ }
+ return db, nil
+}
+
+// UpsertFleet inserts or updates a fleet's account/bucket linkage by name.
+func (r *Registry) UpsertFleet(f Fleet) error {
+ if !ValidTunnelName(f.Name) {
+ return fmt.Errorf("invalid fleet name %q", f.Name)
+ }
+ if f.CreatedAt == 0 {
+ f.CreatedAt = time.Now().Unix()
+ }
+ _, err := r.sql.Exec(`
+INSERT INTO fleets (name, account, bucket, created_at)
+VALUES (?, ?, ?, ?)
+ON CONFLICT(name) DO UPDATE SET
+ account = excluded.account, bucket = excluded.bucket`,
+ f.Name, f.Account, f.Bucket, f.CreatedAt)
+ return err
+}
+
+// GetFleet returns the fleet named name, or (nil, nil) if there is none.
+func (r *Registry) GetFleet(name string) (*Fleet, error) {
+ row := r.sql.QueryRow(`SELECT name, account, bucket, created_at FROM fleets WHERE name = ?`, name)
+ var f Fleet
+ err := row.Scan(&f.Name, &f.Account, &f.Bucket, &f.CreatedAt)
+ if err == sql.ErrNoRows {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &f, nil
+}
+
+// ListFleets returns every fleet, ordered by name, over a fresh connection.
+func (r *Registry) ListFleets() ([]Fleet, error) {
+ db, err := r.freshFleetsConn()
+ if err != nil {
+ return nil, err
+ }
+ defer db.Close()
+ rows, err := db.Query(`SELECT name, account, bucket, created_at FROM fleets ORDER BY name`)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []Fleet
+ for rows.Next() {
+ var f Fleet
+ if err := rows.Scan(&f.Name, &f.Account, &f.Bucket, &f.CreatedAt); err != nil {
+ return nil, err
+ }
+ out = append(out, f)
+ }
+ return out, rows.Err()
+}
+
+// AddFleetBox registers a box to a fleet. It errors if (fleet, label)
+// already names an ACTIVE box — the channel identity must stay unique while
+// live. A label that exists but is DETACHED is reactivated in place with the
+// incoming token/status/added_at, which is the path a re-provisioned box
+// takes back onto its old name.
+func (r *Registry) AddFleetBox(b FleetBox) error {
+ if !ValidTunnelName(b.Fleet) {
+ return fmt.Errorf("invalid fleet name %q", b.Fleet)
+ }
+ if !ValidTunnelName(b.Label) {
+ return fmt.Errorf("invalid box label %q", b.Label)
+ }
+ if b.Status == "" {
+ b.Status = "active"
+ }
+ if b.AddedAt == 0 {
+ b.AddedAt = time.Now().Unix()
+ }
+ tx, err := r.sql.Begin()
+ if err != nil {
+ return err
+ }
+ var status string
+ err = tx.QueryRow(`SELECT status FROM fleet_boxes WHERE fleet = ? AND label = ?`, b.Fleet, b.Label).Scan(&status)
+ if err != nil && err != sql.ErrNoRows {
+ tx.Rollback()
+ return err
+ }
+ if err == nil && status == "active" {
+ tx.Rollback()
+ return fmt.Errorf("fleet %q already has an active box %q", b.Fleet, b.Label)
+ }
+ if _, err := tx.Exec(`
+INSERT INTO fleet_boxes (fleet, label, token_hash, status, added_at)
+VALUES (?, ?, ?, ?, ?)
+ON CONFLICT(fleet, label) DO UPDATE SET
+ token_hash = excluded.token_hash, status = excluded.status, added_at = excluded.added_at`,
+ b.Fleet, b.Label, b.TokenHash, b.Status, b.AddedAt); err != nil {
+ tx.Rollback()
+ return err
+ }
+ return tx.Commit()
+}
+
+// SetFleetBoxToken rotates a box's stored token hash. Errors when no
+// (fleet, label) row exists — a caller minting a fresh token against a box
+// that no longer exists is a race worth reporting, not swallowing.
+func (r *Registry) SetFleetBoxToken(fleet, label, tokenHash string) error {
+ if !ValidTunnelName(fleet) {
+ return fmt.Errorf("invalid fleet name %q", fleet)
+ }
+ if !ValidTunnelName(label) {
+ return fmt.Errorf("invalid box label %q", label)
+ }
+ res, err := r.sql.Exec(`UPDATE fleet_boxes SET token_hash = ? WHERE fleet = ? AND label = ?`, tokenHash, fleet, label)
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n == 0 {
+ return fmt.Errorf("no box %q/%q", fleet, label)
+ }
+ return nil
+}
+
+// SetFleetBoxStatus flips a box between "active" and "detached". Errors when
+// no (fleet, label) row exists, same contract as SetFleetBoxToken.
+func (r *Registry) SetFleetBoxStatus(fleet, label, status string) error {
+ if !ValidTunnelName(fleet) {
+ return fmt.Errorf("invalid fleet name %q", fleet)
+ }
+ if !ValidTunnelName(label) {
+ return fmt.Errorf("invalid box label %q", label)
+ }
+ res, err := r.sql.Exec(`UPDATE fleet_boxes SET status = ? WHERE fleet = ? AND label = ?`, status, fleet, label)
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n == 0 {
+ return fmt.Errorf("no box %q/%q", fleet, label)
+ }
+ return nil
+}
+
+// ListFleetBoxes returns every box registered to fleet, ordered by label,
+// over a fresh connection (same rationale as ListFleets).
+func (r *Registry) ListFleetBoxes(fleet string) ([]FleetBox, error) {
+ db, err := r.freshFleetsConn()
+ if err != nil {
+ return nil, err
+ }
+ defer db.Close()
+ rows, err := db.Query(`SELECT fleet, label, token_hash, status, added_at FROM fleet_boxes WHERE fleet = ? ORDER BY label`, fleet)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []FleetBox
+ for rows.Next() {
+ var b FleetBox
+ if err := rows.Scan(&b.Fleet, &b.Label, &b.TokenHash, &b.Status, &b.AddedAt); err != nil {
+ return nil, err
+ }
+ out = append(out, b)
+ }
+ return out, rows.Err()
+}
+
+// FleetChannelAuth returns every ACTIVE box's token hash, keyed by its
+// channel name "<fleet>/<label>" — the edge's auth table for an incoming
+// remote-host-agent dial-in. Fresh connection: this is exactly the
+// cross-process poll List/Fingerprint exist for (the edge holds a long-lived
+// Registry handle and reloads this whenever Fingerprint moves).
+func (r *Registry) FleetChannelAuth() (map[string]string, error) {
+ db, err := r.freshFleetsConn()
+ if err != nil {
+ return nil, err
+ }
+ defer db.Close()
+ rows, err := db.Query(`SELECT fleet, label, token_hash FROM fleet_boxes WHERE status = 'active'`)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ out := make(map[string]string)
+ for rows.Next() {
+ var fleet, label, hash string
+ if err := rows.Scan(&fleet, &label, &hash); err != nil {
+ return nil, err
+ }
+ out[fleet+"/"+label] = hash
+ }
+ return out, rows.Err()
+}
diff --git a/internal/registry/fleets_test.go b/internal/registry/fleets_test.go
new file mode 100644
index 000000000..777369756
--- /dev/null
+++ b/internal/registry/fleets_test.go
@@ -0,0 +1,205 @@
+package registry
+
+import "testing"
+
+// TestFleetRoundTrips pins UpsertFleet/GetFleet/ListFleets against the exact
+// shape later tasks depend on (task-1-brief.md's Fleet struct).
+func TestFleetRoundTrips(t *testing.T) {
+ reg := openTestRegistry(t)
+ want := Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data", CreatedAt: 1700000000}
+ if err := reg.UpsertFleet(want); err != nil {
+ t.Fatal(err)
+ }
+ got, err := reg.GetFleet("acme")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got == nil {
+ t.Fatal("GetFleet returned nil for an upserted fleet")
+ }
+ if *got != want {
+ t.Fatalf("round trip mismatch:\n got %+v\n want %+v", *got, want)
+ }
+ list, err := reg.ListFleets()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(list) != 1 || list[0] != want {
+ t.Fatalf("ListFleets = %+v, want [%+v]", list, want)
+ }
+}
+
+func TestGetFleetMissingIsNilNil(t *testing.T) {
+ reg := openTestRegistry(t)
+ got, err := reg.GetFleet("nope")
+ if err != nil || got != nil {
+ t.Fatalf("GetFleet(missing) = %v, %v; want nil, nil", got, err)
+ }
+}
+
+// TestFleetBoxRoundTrips pins AddFleetBox/ListFleetBoxes against the exact
+// FleetBox shape later tasks depend on.
+func TestFleetBoxRoundTrips(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.UpsertFleet(Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil {
+ t.Fatal(err)
+ }
+ want := FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: "deadbeef", Status: "active", AddedAt: 1700000001}
+ if err := reg.AddFleetBox(want); err != nil {
+ t.Fatal(err)
+ }
+ boxes, err := reg.ListFleetBoxes("acme")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(boxes) != 1 || boxes[0] != want {
+ t.Fatalf("ListFleetBoxes = %+v, want [%+v]", boxes, want)
+ }
+}
+
+// TestAddFleetBoxRejectsDuplicateActive is the brief's core invariant: two
+// boxes cannot hold the same (fleet, label) channel identity while both are
+// active.
+func TestAddFleetBoxRejectsDuplicateActive(t *testing.T) {
+ reg := openTestRegistry(t)
+ box := FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: "hash1", Status: "active"}
+ if err := reg.AddFleetBox(box); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.AddFleetBox(box); err == nil {
+ t.Fatal("AddFleetBox on a duplicate active (fleet,label): got nil error, want one")
+ }
+}
+
+// TestAddFleetBoxReactivatesADetachedLabel: the brief only forbids a
+// duplicate ACTIVE box, implying a detached label can be re-added — the path
+// a re-provisioned box takes.
+func TestAddFleetBoxReactivatesADetachedLabel(t *testing.T) {
+ reg := openTestRegistry(t)
+ box := FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: "hash1", Status: "active"}
+ if err := reg.AddFleetBox(box); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetFleetBoxStatus("acme", "pi-1", "detached"); err != nil {
+ t.Fatal(err)
+ }
+ reAdd := FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: "hash2", Status: "active"}
+ if err := reg.AddFleetBox(reAdd); err != nil {
+ t.Fatalf("AddFleetBox over a detached label: %v", err)
+ }
+ boxes, err := reg.ListFleetBoxes("acme")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(boxes) != 1 || boxes[0].TokenHash != "hash2" || boxes[0].Status != "active" {
+ t.Fatalf("re-add over a detached label = %+v, want reactivated with the new token", boxes)
+ }
+}
+
+// TestFleetChannelAuthOmitsDetachedBoxes pins the exact contract later tasks
+// rely on: channel name "<fleet>/<label>" -> TokenHash, ACTIVE boxes only.
+func TestFleetChannelAuthOmitsDetachedBoxes(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.AddFleetBox(FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: "hash1", Status: "active"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.AddFleetBox(FleetBox{Fleet: "acme", Label: "pi-2", TokenHash: "hash2", Status: "active"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetFleetBoxStatus("acme", "pi-2", "detached"); err != nil {
+ t.Fatal(err)
+ }
+ auth, err := reg.FleetChannelAuth()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, ok := auth["acme/pi-2"]; ok {
+ t.Fatalf("FleetChannelAuth included a detached box: %+v", auth)
+ }
+ if got, ok := auth["acme/pi-1"]; !ok || got != "hash1" {
+ t.Fatalf("FleetChannelAuth missing/wrong entry for the active box: %+v", auth)
+ }
+}
+
+func TestSetFleetBoxTokenMissingRowErrors(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.SetFleetBoxToken("acme", "nope", "hash"); err == nil {
+ t.Fatal("SetFleetBoxToken on a missing row: got nil error, want one")
+ }
+}
+
+func TestSetFleetBoxStatusMissingRowErrors(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.SetFleetBoxStatus("acme", "nope", "detached"); err == nil {
+ t.Fatal("SetFleetBoxStatus on a missing row: got nil error, want one")
+ }
+}
+
+// TestFleetWritersRejectInvalidNames: every writer validates fleet/label
+// names with ValidTunnelName before touching the DB.
+func TestFleetWritersRejectInvalidNames(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.UpsertFleet(Fleet{Name: "UPPER"}); err == nil {
+ t.Fatal("UpsertFleet({Name:UPPER}): got nil error, want one")
+ }
+ if err := reg.AddFleetBox(FleetBox{Fleet: "UPPER", Label: "pi-1"}); err == nil {
+ t.Fatal("AddFleetBox with an invalid fleet name: got nil error, want one")
+ }
+ if err := reg.AddFleetBox(FleetBox{Fleet: "acme", Label: "UPPER"}); err == nil {
+ t.Fatal("AddFleetBox with an invalid label: got nil error, want one")
+ }
+ if err := reg.SetFleetBoxToken("UPPER", "pi-1", "hash"); err == nil {
+ t.Fatal("SetFleetBoxToken with an invalid fleet name: got nil error, want one")
+ }
+ if err := reg.SetFleetBoxStatus("acme", "UPPER", "detached"); err == nil {
+ t.Fatal("SetFleetBoxStatus with an invalid label: got nil error, want one")
+ }
+}
+
+// TestFingerprintChangesOnFleetMutation: Fingerprint must fold the fleets
+// table into its aggregate — mutating a FLEET row (not a route) must still
+// change it, which is what lets the edge notice a fleet change later.
+func TestFingerprintChangesOnFleetMutation(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.UpsertFleet(Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil {
+ t.Fatal(err)
+ }
+ before, err := reg.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.UpsertFleet(Fleet{Name: "acme", Account: "amadan-longer-name", Bucket: "acme-data"}); err != nil {
+ t.Fatal(err)
+ }
+ after, err := reg.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if before == after {
+ t.Fatalf("fingerprint unchanged across a fleet mutation: %q", after)
+ }
+}
+
+// TestFingerprintChangesOnFleetBoxMutation: same guarantee for fleet_boxes —
+// a box status flip must change the fingerprint too (SetFleetBoxStatus is
+// exactly how a detach reaches the edge in later tasks).
+func TestFingerprintChangesOnFleetBoxMutation(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.AddFleetBox(FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: "hash1", Status: "active"}); err != nil {
+ t.Fatal(err)
+ }
+ before, err := reg.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetFleetBoxStatus("acme", "pi-1", "detached"); err != nil {
+ t.Fatal(err)
+ }
+ after, err := reg.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if before == after {
+ t.Fatalf("fingerprint unchanged across a fleet_boxes status flip: %q", after)
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 797b08494..07951f3c9 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -261,7 +261,10 @@ CREATE TABLE IF NOT EXISTS routes (
return err
}
}
- return r.migrateAccounts()
+ if err := r.migrateAccounts(); err != nil {
+ return err
+ }
+ return r.migrateFleets()
}
// migrateAccounts brings the accounts table to its multi-account shape:
@@ -983,10 +986,11 @@ WHERE kind = 'instance' AND next_due > 0 AND next_due <= ? ORDER BY host`, now)
return out, rows.Err()
}
-// Fingerprint summarizes the route table over a fresh connection, for the
-// edge's cheap change poll. It must change on any row insert/delete and on
-// the column writes reloads care about; it must see other processes' commits
-// (fresh connection, same WAL-snapshot quirk as List).
+// Fingerprint summarizes the routes, fleets and fleet_boxes tables over a
+// fresh connection, for the edge's cheap change poll. It must change on any
+// row insert/delete and on the column writes reloads care about; it must see
+// other processes' commits (fresh connection, same WAL-snapshot quirk as
+// List).
//
// Text columns contribute their LENGTH under a per-column weight, so a rewrite
// to a DIFFERENT value of the SAME length (channel canary/1 -> canary/2,
@@ -996,6 +1000,14 @@ WHERE kind = 'instance' AND next_due > 0 AND next_due <= ? ORDER BY host`, now)
// the fleet on the next reload for any other reason. Weights (3,5,7,11,13…)
// keep DIFFERENT columns from cancelling each other out, which is the failure
// that would actually hide a change forever.
+//
+// fleets and fleet_boxes each get their own weighted scalar subquery (own
+// weight sets — 19/23/29 and 31/37/41/43/47 — so a routes change and a
+// fleets change can never cancel each other out either), concatenated onto
+// the routes aggregate: still ONE query, so a fleet or fleet_boxes mutation
+// (an account move, a box's status flip, a token rotation) changes the
+// fingerprint exactly like a route write does, which is what lets the edge's
+// Watch poll (edge.go:830) notice a fleet change and reload.
func (r *Registry) Fingerprint() (string, error) {
db, err := sql.Open("sqlite", r.path)
if err != nil {
@@ -1007,10 +1019,21 @@ func (r *Registry) Fingerprint() (string, error) {
}
var fp string
err = db.QueryRow(`
-SELECT COUNT(*) || ':' || COALESCE(MAX(rowid),0) || ':' ||
- COALESCE(SUM(rowid*17 + LENGTH(status)*3 + LENGTH(socket) + LENGTH(addr)
- + LENGTH(version)*5 + hibernate*7 + LENGTH(channel)
- + LENGTH(backing)*11 + LENGTH(tunnel)*13),0)
-FROM routes`).Scan(&fp)
+SELECT
+ (SELECT COUNT(*) || ':' || COALESCE(MAX(rowid),0) || ':' ||
+ COALESCE(SUM(rowid*17 + LENGTH(status)*3 + LENGTH(socket) + LENGTH(addr)
+ + LENGTH(version)*5 + hibernate*7 + LENGTH(channel)
+ + LENGTH(backing)*11 + LENGTH(tunnel)*13),0)
+ FROM routes)
+ || ':' ||
+ (SELECT COUNT(*) || ':' || COALESCE(MAX(rowid),0) || ':' ||
+ COALESCE(SUM(rowid*19 + LENGTH(account)*23 + LENGTH(bucket)*29),0)
+ FROM fleets)
+ || ':' ||
+ (SELECT COUNT(*) || ':' || COALESCE(MAX(rowid),0) || ':' ||
+ COALESCE(SUM(rowid*31 + LENGTH(fleet)*37 + LENGTH(label)*41
+ + LENGTH(token_hash)*43 + LENGTH(status)*47),0)
+ FROM fleet_boxes)
+`).Scan(&fp)
return fp, err
}