| index b498b583f..2e2e17469 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -36,6 +36,21 @@ type Route struct { |
| Kind string // instance | service | static |
| Socket string // unix socket upstream; wins over Addr when both set |
| Addr string // TCP upstream host:port — whole-app tenants (Kass v1) |
| + // Addrs is a route's multi-upstream set (HA design, issue #152): a |
| + // comma-joined list of TCP upstream host:port entries, "" = none. The |
| + // comma-join is safe because every element is validated by |
| + // provision.validAddr, whose shape (host:port) cannot itself contain a |
| + // comma. Addr above remains the legacy single-upstream field — it is |
| + // untouched by multi-upstream tenants and AddrList folds it into a |
| + // one-element list when Addrs is empty, so a single-upstream caller sees |
| + // no difference. |
| + Addrs string |
| + // Health is the health-probe config JSON, verbatim, from the record that |
| + // minted this route — the canonical json.Marshal of provision.Health |
| + // (wire key "probe" on the record; the rename there was JSON-only, so |
| + // this column keeps the internal name). "" means unprobed, not a zero or |
| + // failing probe. |
| + Health string |
| Status string // pending|creating|starting|issuing|ready|failed |
| |
| LastError string |
| @@ -191,6 +206,22 @@ type Route struct { |
| CreatedAt int64 |
| } |
| |
| +// AddrList resolves a route's upstream set for a multi-upstream-aware caller: |
| +// Addrs split on the comma when it is set, else Addr folded into a |
| +// one-element list for the legacy single-upstream shape, else nil when the |
| +// route names no TCP upstream at all (a socket-backed or static route). The |
| +// nil case matters to callers that branch on "has upstreams" — an empty |
| +// non-nil slice would read as "has zero upstreams", a different fact. |
| +func (t Route) AddrList() []string { |
| + if t.Addrs != "" { |
| + return strings.Split(t.Addrs, ",") |
| + } |
| + if t.Addr != "" { |
| + return []string{t.Addr} |
| + } |
| + return nil |
| +} |
| + |
| // BackingUnit marks a route whose process is a carlos-app@<host> systemd unit |
| // rather than an exec child of the agent. The empty string stays the exec |
| // default so pre-existing rows keep today's behavior byte-identical. |
| @@ -328,6 +359,8 @@ CREATE TABLE IF NOT EXISTS routes ( |
| `ALTER TABLE routes ADD COLUMN seed TEXT NOT NULL DEFAULT ''`, |
| `ALTER TABLE routes ADD COLUMN unit TEXT NOT NULL DEFAULT ''`, |
| `ALTER TABLE routes ADD COLUMN fleet_box TEXT NOT NULL DEFAULT ''`, |
| + `ALTER TABLE routes ADD COLUMN addrs TEXT NOT NULL DEFAULT ''`, |
| + `ALTER TABLE routes ADD COLUMN health TEXT NOT NULL DEFAULT ''`, |
| } { |
| if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| return err |
| @@ -494,7 +527,7 @@ func (r *Registry) Accounts() ([]Account, error) { |
| } |
| |
| // routeCols is the column list every route read and the one route write |
| -// share — 33 names, in this exact order. scanRoute's destinations, Upsert's |
| +// share — 35 names, in this exact order. scanRoute's destinations, Upsert's |
| // placeholder row and Upsert's argument list all count from here: adding a |
| // column means adding it in FOUR places, and a miscount is a silent column |
| // shift rather than an error (TestUpsertRoundTripsAllColumns is what catches |
| @@ -503,7 +536,7 @@ const routeCols = `host, slug, kind, socket, addr, status, last_error, db_path, |
| version, prev_version, app, pool, region, channel, hibernate, canary, |
| replication, next_due, backing, status_named, platform, dns_status, |
| dns_checked_at, cert_expires_at, account, environment, caps, tunnel, |
| -provisioned, seed, unit, fleet_box, created_at` |
| +provisioned, seed, unit, fleet_box, addrs, health, created_at` |
| |
| func scanRoute(scan func(...any) error) (Route, error) { |
| var t Route |
| @@ -513,7 +546,7 @@ func scanRoute(scan func(...any) error) (Route, error) { |
| &t.Region, &t.Channel, &hib, &can, &t.Replication, &t.NextDue, &t.Backing, |
| &sn, &plat, &t.DNSStatus, &t.DNSCheckedAt, &t.CertExpiresAt, &t.Account, |
| &t.Environment, &t.Caps, &t.Tunnel, &prov, &t.Seed, &t.Unit, &t.FleetBox, |
| - &t.CreatedAt) |
| + &t.Addrs, &t.Health, &t.CreatedAt) |
| t.Hibernate = hib != 0 |
| t.Canary = can != 0 |
| t.StatusNamed = sn != 0 |
| @@ -628,7 +661,7 @@ func (r *Registry) Upsert(t Route) error { |
| } |
| _, err := r.sql.Exec(` |
| INSERT INTO routes (`+routeCols+`) |
| -VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| +VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| ON CONFLICT(host) DO UPDATE SET |
| slug=excluded.slug, kind=excluded.kind, socket=excluded.socket, |
| addr=excluded.addr, status=excluded.status, last_error=excluded.last_error, |
| @@ -647,13 +680,14 @@ ON CONFLICT(host) DO UPDATE SET |
| provisioned=CASE WHEN excluded.provisioned = 0 THEN routes.provisioned ELSE excluded.provisioned END, |
| seed=CASE WHEN excluded.seed = '' THEN routes.seed ELSE excluded.seed END, |
| unit=CASE WHEN excluded.unit = '' THEN routes.unit ELSE excluded.unit END, |
| - fleet_box=CASE WHEN excluded.fleet_box = '' THEN routes.fleet_box ELSE excluded.fleet_box END`, |
| + fleet_box=CASE WHEN excluded.fleet_box = '' THEN routes.fleet_box ELSE excluded.fleet_box END, |
| + addrs=excluded.addrs, health=excluded.health`, |
| t.Host, t.Slug, t.Kind, t.Socket, t.Addr, t.Status, t.LastError, |
| t.DBPath, t.Version, t.PrevVersion, t.App, t.Pool, t.Region, t.Channel, |
| b2i(t.Hibernate), b2i(t.Canary), t.Replication, t.NextDue, t.Backing, |
| b2i(t.StatusNamed), b2i(t.Platform), t.DNSStatus, t.DNSCheckedAt, |
| t.CertExpiresAt, t.Account, t.Environment, t.Caps, t.Tunnel, |
| - b2i(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.CreatedAt) |
| + b2i(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health, t.CreatedAt) |
| return err |
| } |
| |
| @@ -858,6 +892,35 @@ func (r *Registry) SetUnit(host, unit string) error { |
| return nil |
| } |
| |
| +// SetUpstreams repoints a route's multi-upstream set as a narrow UPDATE — |
| +// same rationale as SetChannel/SetUnit: a read-modify-write Upsert racing a |
| +// deploy's SetVersions or the activator's SetNextDue would clobber those |
| +// columns, and the provision reconcile pass (the only writer of these two |
| +// columns today) converges them independently of everything else a full-row |
| +// Upsert would touch. addrs is the caller's already comma-joined list ("" = |
| +// none); health is the probe-config JSON verbatim ("" = unprobed). Neither |
| +// argument is validated here — the rails live at the provision pass's own |
| +// door (ValidateShape), which runs before a record ever reaches this column, |
| +// exactly as SetCaps's callers validate before calling it. |
| +// |
| +// Errors on an unmatched host, the SetChannel/SetUnit contract: a silent |
| +// no-op on a host that does not exist would hide a reconcile pass writing to |
| +// the wrong row. |
| +func (r *Registry) SetUpstreams(host, addrs, health string) error { |
| + res, err := r.sql.Exec(`UPDATE routes SET addrs = ?, health = ? WHERE host = ?`, addrs, health, host) |
| + if err != nil { |
| + return err |
| + } |
| + n, err := res.RowsAffected() |
| + if err != nil { |
| + return err |
| + } |
| + if n == 0 { |
| + return fmt.Errorf("no route for host %q", host) |
| + } |
| + return nil |
| +} |
| + |
| // SetChannels repoints every host in hosts to channel as ONE transaction — |
| // every row moves or none do. `carlos route -app` uses this for its bulk |
| // form: the caller runs every route's safety-rail check to completion FIRST |
| @@ -1246,6 +1309,12 @@ WHERE kind = 'instance' AND next_due > 0 AND next_due <= ? ORDER BY host`, now) |
| // database must reach a running edge, or the activator keeps waking the host |
| // on the stale cached path. |
| // |
| +// `addrs` and `health` are hashed too (issue #152, multi-upstream routing): a |
| +// SetUpstreams call that resizes or reshuffles a route's upstream set, or |
| +// changes its probe config, is exactly as reload-relevant as an addr rewrite |
| +// — an edge that never noticed would keep dispatching to a dropped upstream |
| +// or never start probing a newly-added one. |
| +// |
| // `region` and `fleet_box` are hashed for the customer-fleets placement path, |
| // and the two-box e2e (internal/edge/fleet_e2e_test.go) is what proved they |
| // have to be: a failover restamp (ReconcileFleetBindings -> SetFleetBox, |
| @@ -1284,7 +1353,7 @@ func (r *Registry) Fingerprint() (string, error) { |
| } |
| rows, err := db.Query(` |
| SELECT rowid, status, socket, addr, version, hibernate, channel, backing, tunnel, |
| - db_path, region, fleet_box |
| + db_path, region, fleet_box, addrs, health |
| FROM routes ORDER BY rowid`) |
| if err != nil { |
| return "", err |
| @@ -1294,14 +1363,14 @@ FROM routes ORDER BY rowid`) |
| n := 0 |
| for rows.Next() { |
| var rowid, hibernate int64 |
| - var status, socket, addr, version, channel, backing, tunnel, dbPath, region, fleetBox string |
| + var status, socket, addr, version, channel, backing, tunnel, dbPath, region, fleetBox, addrs, health string |
| if err := rows.Scan(&rowid, &status, &socket, &addr, &version, &hibernate, |
| - &channel, &backing, &tunnel, &dbPath, ®ion, &fleetBox); err != nil { |
| + &channel, &backing, &tunnel, &dbPath, ®ion, &fleetBox, &addrs, &health); err != nil { |
| return "", err |
| } |
| - fmt.Fprintf(h, "%d\x1f%s\x1f%s\x1f%s\x1f%s\x1f%d\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1e", |
| + fmt.Fprintf(h, "%d\x1f%s\x1f%s\x1f%s\x1f%s\x1f%d\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1e", |
| rowid, status, socket, addr, version, hibernate, channel, backing, tunnel, |
| - dbPath, region, fleetBox) |
| + dbPath, region, fleetBox, addrs, health) |
| n++ |
| } |
| if err := rows.Err(); err != nil { |