| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "database/sql" |
| 5 | "fmt" |
| 6 | "strings" |
| 7 | "time" |
| 8 | ) |
| 9 | |
| 10 | // Registry is the router's control DB — the set of hosts it serves and how to |
| 11 | // reach each. It's deliberately separate from an instance's own SQLite: it holds |
| 12 | // routing + operator config, never message data. See docs/instance-router.md. |
| 13 | type Registry struct{ sql *sql.DB } |
| 14 | |
| 15 | // Route is one host the router fronts: a messenger instance it lifecycle-manages |
| 16 | // (kind "instance") or a sibling service it only proxies (kind "service", e.g. |
| 17 | // the home server or push relay). The instance-only fields hold per-instance |
| 18 | // config that used to live in process env. |
| 19 | type Route struct { |
| 20 | Host string |
| 21 | Kind string |
| 22 | Socket string |
| 23 | Status string |
| 24 | LastError string |
| 25 | DBPath string |
| 26 | Version string |
| 27 | PrevVersion string |
| 28 | AppStoreURL string |
| 29 | HomeOrigin string |
| 30 | PushRelayURL string |
| 31 | PushRelayToken string |
| 32 | // HomeRegion + Provider are the geo and co-op seams (#550 Parts 7 + 10): |
| 33 | // where a space lives and who runs it. Schema-only today — "" means the |
| 34 | // single implicit region/provider (this box), and nothing keys behavior on |
| 35 | // them yet; register() just records what a caller sets. |
| 36 | HomeRegion string |
| 37 | Provider string |
| 38 | // Hibernate is the per-host opt-in to activator management (#550 Part 4 |
| 39 | // v2): with the router's LCHAT_HIBERNATE_BUCKET set, a flagged instance |
| 40 | // runs as an on-demand child — woken per request, slept when idle — instead |
| 41 | // of an always-on systemd unit. Off (the default) keeps the classic systemd |
| 42 | // shape even when the router has the env set, so one production canary can |
| 43 | // hibernate while the rest of the box is untouched. Flipped by |
| 44 | // `lchat hibernate <host> --on/--off` via setHibernate. |
| 45 | Hibernate bool |
| 46 | // Canary marks a throwaway instance (a per-session test host) — the OPPOSITE |
| 47 | // of member data. It's set deliberately (add-instance --canary, or |
| 48 | // `lchat canary <host> --on`), NEVER inferred from the hostname: test.luan.chat |
| 49 | // and test-bots.luan.chat are long-lived data-holding instances despite the |
| 50 | // prefix, so name-matching would be dangerously wrong. The signal exists so a |
| 51 | // future scoped reset/wipe can refuse anything not flagged, and so fleet ops |
| 52 | // and status can treat canaries distinctly. |
| 53 | Canary bool |
| 54 | // Replication is the host's durability class on the activator's S3 replica |
| 55 | // — the per-plan ladder ("stream"/"batched"/"daily"; "" = stream, the |
| 56 | // historical behavior). It sets how often litestream's replica syncs local |
| 57 | // changes to S3 while the host is awake; every graceful path (sleep, |
| 58 | // deploy, shutdown) still flushes everything regardless of class, so the |
| 59 | // class only bounds what a HARD crash of the box can lose. Inert for |
| 60 | // systemd-managed hosts (no activator replication to tune). Applied at |
| 61 | // wake time — a change takes effect on the next wake. Flipped by |
| 62 | // `lchat replication <host> <class>` via setReplication. |
| 63 | Replication string |
| 64 | // NextDue is the due-time index (#550 Part 4 v2): the earliest pending |
| 65 | // scheduled delivery (unix seconds; 0 = none) recorded by the activator when |
| 66 | // the instance hibernates, so the sweeper can wake it in time to fire. In |
| 67 | // the registry — not activator memory — so due times survive router restarts |
| 68 | // (the registry is fleet-backed-up). Written via setNextDue only. |
| 69 | NextDue int64 |
| 70 | CreatedAt int64 |
| 71 | } |
| 72 | |
| 73 | // The replication classes — the durability ladder a plan buys (#853). Class |
| 74 | // names describe the mechanism, not the plan, so billing can remap tiers to |
| 75 | // classes without touching the router. All classes flush in full on every |
| 76 | // graceful sleep/deploy/shutdown; the sync interval below only bounds the |
| 77 | // hard-crash loss window (and its S3 Tier-1 request bill) while awake. |
| 78 | const ( |
| 79 | replicationStream = "stream" // change-driven ~1s sync — loss window seconds |
| 80 | replicationBatched = "batched" // sync at most every 5m — loss window minutes |
| 81 | replicationDaily = "daily" // sync at wake + every 24h — loss window up to a day |
| 82 | ) |
| 83 | |
| 84 | // validReplication reports whether s names a replication class ("" counts: |
| 85 | // it's the unset default, stream). |
| 86 | func validReplication(s string) bool { |
| 87 | switch s { |
| 88 | case "", replicationStream, replicationBatched, replicationDaily: |
| 89 | return true |
| 90 | } |
| 91 | return false |
| 92 | } |
| 93 | |
| 94 | // replicationSyncInterval maps a class to the litestream replica sync |
| 95 | // interval it buys. 0 means "keep litestream's default" (1s, change-driven — |
| 96 | // idle ticks make zero S3 calls), which is what stream and the unset "" |
| 97 | // deliberately share so pre-ladder registries keep today's behavior exactly. |
| 98 | func replicationSyncInterval(class string) time.Duration { |
| 99 | switch class { |
| 100 | case replicationBatched: |
| 101 | return 5 * time.Minute |
| 102 | case replicationDaily: |
| 103 | return 24 * time.Hour |
| 104 | } |
| 105 | return 0 |
| 106 | } |
| 107 | |
| 108 | func openRegistry(path string) (*Registry, error) { |
| 109 | d, err := sql.Open("sqlite", path) |
| 110 | if err != nil { |
| 111 | return nil, err |
| 112 | } |
| 113 | // One writer at a time keeps WAL happy (mirrors openDB). |
| 114 | d.SetMaxOpenConns(1) |
| 115 | // busy_timeout FIRST: two processes open this DB concurrently (the router |
| 116 | // starting while add-instance or the tunnel gateway runs), and |
| 117 | // journal_mode=WAL takes a lock — with no timeout set yet the loser fails |
| 118 | // immediately with SQLITE_BUSY instead of waiting its turn. |
| 119 | for _, p := range []string{ |
| 120 | "PRAGMA busy_timeout=5000", |
| 121 | "PRAGMA journal_mode=WAL", |
| 122 | "PRAGMA synchronous=NORMAL", |
| 123 | } { |
| 124 | if _, err := d.Exec(p); err != nil { |
| 125 | return nil, err |
| 126 | } |
| 127 | } |
| 128 | r := &Registry{sql: d} |
| 129 | if err := r.migrate(); err != nil { |
| 130 | return nil, err |
| 131 | } |
| 132 | return r, nil |
| 133 | } |
| 134 | |
| 135 | // migrate is additive (CREATE TABLE IF NOT EXISTS), like the instance schema, so |
| 136 | // deploying new code over an existing registry is safe. |
| 137 | func (r *Registry) migrate() error { |
| 138 | if _, err := r.sql.Exec(` |
| 139 | CREATE TABLE IF NOT EXISTS routes ( |
| 140 | host TEXT PRIMARY KEY, |
| 141 | kind TEXT NOT NULL, |
| 142 | socket TEXT NOT NULL, |
| 143 | status TEXT NOT NULL DEFAULT 'ready', |
| 144 | last_error TEXT NOT NULL DEFAULT '', |
| 145 | db_path TEXT NOT NULL DEFAULT '', |
| 146 | version TEXT NOT NULL DEFAULT '', |
| 147 | prev_version TEXT NOT NULL DEFAULT '', |
| 148 | app_store_url TEXT NOT NULL DEFAULT '', |
| 149 | home_origin TEXT NOT NULL DEFAULT '', |
| 150 | push_relay_url TEXT NOT NULL DEFAULT '', |
| 151 | push_relay_token TEXT NOT NULL DEFAULT '', |
| 152 | home_region TEXT NOT NULL DEFAULT '', |
| 153 | provider TEXT NOT NULL DEFAULT '', |
| 154 | hibernate INTEGER NOT NULL DEFAULT 0, |
| 155 | canary INTEGER NOT NULL DEFAULT 0, |
| 156 | replication TEXT NOT NULL DEFAULT '', |
| 157 | next_due INTEGER NOT NULL DEFAULT 0, |
| 158 | created_at INTEGER NOT NULL |
| 159 | );`); err != nil { |
| 160 | return err |
| 161 | } |
| 162 | // Additive column migrations for registries created before a column existed — |
| 163 | // CREATE TABLE IF NOT EXISTS won't add columns to an existing table. Ignore the |
| 164 | // "duplicate column" error so this stays idempotent. |
| 165 | for _, stmt := range []string{ |
| 166 | `ALTER TABLE routes ADD COLUMN prev_version TEXT NOT NULL DEFAULT ''`, |
| 167 | `ALTER TABLE routes ADD COLUMN home_region TEXT NOT NULL DEFAULT ''`, |
| 168 | `ALTER TABLE routes ADD COLUMN provider TEXT NOT NULL DEFAULT ''`, |
| 169 | `ALTER TABLE routes ADD COLUMN next_due INTEGER NOT NULL DEFAULT 0`, |
| 170 | `ALTER TABLE routes ADD COLUMN hibernate INTEGER NOT NULL DEFAULT 0`, |
| 171 | `ALTER TABLE routes ADD COLUMN replication TEXT NOT NULL DEFAULT ''`, |
| 172 | `ALTER TABLE routes ADD COLUMN canary INTEGER NOT NULL DEFAULT 0`, |
| 173 | } { |
| 174 | if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| 175 | return err |
| 176 | } |
| 177 | } |
| 178 | return nil |
| 179 | } |
| 180 | |
| 181 | // list returns every route, ordered by host. |
| 182 | func (r *Registry) list() ([]Route, error) { |
| 183 | rows, err := r.sql.Query(` |
| 184 | SELECT host, kind, socket, status, last_error, db_path, version, prev_version, |
| 185 | app_store_url, home_origin, push_relay_url, push_relay_token, |
| 186 | home_region, provider, hibernate, canary, replication, next_due, created_at |
| 187 | FROM routes ORDER BY host`) |
| 188 | if err != nil { |
| 189 | return nil, err |
| 190 | } |
| 191 | defer rows.Close() |
| 192 | var out []Route |
| 193 | for rows.Next() { |
| 194 | var t Route |
| 195 | if err := rows.Scan(&t.Host, &t.Kind, &t.Socket, &t.Status, &t.LastError, |
| 196 | &t.DBPath, &t.Version, &t.PrevVersion, &t.AppStoreURL, &t.HomeOrigin, |
| 197 | &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.Hibernate, &t.Canary, &t.Replication, &t.NextDue, &t.CreatedAt); err != nil { |
| 198 | return nil, err |
| 199 | } |
| 200 | out = append(out, t) |
| 201 | } |
| 202 | return out, rows.Err() |
| 203 | } |
| 204 | |
| 205 | // get returns the route for host, or (nil, nil) if there is none. |
| 206 | func (r *Registry) get(host string) (*Route, error) { |
| 207 | var t Route |
| 208 | err := r.sql.QueryRow(` |
| 209 | SELECT host, kind, socket, status, last_error, db_path, version, prev_version, |
| 210 | app_store_url, home_origin, push_relay_url, push_relay_token, |
| 211 | home_region, provider, hibernate, canary, replication, next_due, created_at |
| 212 | FROM routes WHERE host = ?`, host).Scan(&t.Host, &t.Kind, &t.Socket, &t.Status, |
| 213 | &t.LastError, &t.DBPath, &t.Version, &t.PrevVersion, &t.AppStoreURL, &t.HomeOrigin, |
| 214 | &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.Hibernate, &t.Canary, &t.Replication, &t.NextDue, &t.CreatedAt) |
| 215 | if err == sql.ErrNoRows { |
| 216 | return nil, nil |
| 217 | } |
| 218 | if err != nil { |
| 219 | return nil, err |
| 220 | } |
| 221 | return &t, nil |
| 222 | } |
| 223 | |
| 224 | // delete removes a route by host (data-safe: it stops routing the host but never |
| 225 | // touches its DB or env file). |
| 226 | func (r *Registry) delete(host string) error { |
| 227 | _, err := r.sql.Exec(`DELETE FROM routes WHERE host = ?`, host) |
| 228 | return err |
| 229 | } |
| 230 | |
| 231 | // upsert inserts or updates a route by host. (add-instance / register will drive |
| 232 | // this in a later step; it's here now so the router and its tests can seed |
| 233 | // routes.) |
| 234 | func (r *Registry) upsert(t Route) error { |
| 235 | if t.CreatedAt == 0 { |
| 236 | t.CreatedAt = time.Now().Unix() |
| 237 | } |
| 238 | if t.Status == "" { |
| 239 | t.Status = "ready" |
| 240 | } |
| 241 | if t.Kind == "" { |
| 242 | t.Kind = "instance" |
| 243 | } |
| 244 | _, err := r.sql.Exec(` |
| 245 | INSERT INTO routes (host, kind, socket, status, last_error, db_path, version, prev_version, |
| 246 | app_store_url, home_origin, push_relay_url, push_relay_token, |
| 247 | home_region, provider, hibernate, canary, replication, next_due, created_at) |
| 248 | VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| 249 | ON CONFLICT(host) DO UPDATE SET |
| 250 | kind=excluded.kind, socket=excluded.socket, status=excluded.status, |
| 251 | last_error=excluded.last_error, db_path=excluded.db_path, version=excluded.version, |
| 252 | prev_version=excluded.prev_version, |
| 253 | app_store_url=excluded.app_store_url, home_origin=excluded.home_origin, |
| 254 | push_relay_url=excluded.push_relay_url, push_relay_token=excluded.push_relay_token, |
| 255 | home_region=excluded.home_region, provider=excluded.provider, |
| 256 | hibernate=excluded.hibernate, canary=excluded.canary, replication=excluded.replication, next_due=excluded.next_due`, |
| 257 | t.Host, t.Kind, t.Socket, t.Status, t.LastError, t.DBPath, t.Version, t.PrevVersion, |
| 258 | t.AppStoreURL, t.HomeOrigin, t.PushRelayURL, t.PushRelayToken, |
| 259 | t.HomeRegion, t.Provider, t.Hibernate, t.Canary, t.Replication, t.NextDue, t.CreatedAt) |
| 260 | return err |
| 261 | } |
| 262 | |
| 263 | // setHibernate flips host's activator-managed flag without touching the rest of |
| 264 | // the route — a narrow UPDATE (like setNextDue) so the flip can never clobber a |
| 265 | // concurrent next_due write or a deploy's read-modify-write, and vice versa. An |
| 266 | // unknown host is a quiet no-op, matching setNextDue. |
| 267 | func (r *Registry) setHibernate(host string, on bool) error { |
| 268 | _, err := r.sql.Exec(`UPDATE routes SET hibernate = ? WHERE host = ?`, on, host) |
| 269 | return err |
| 270 | } |
| 271 | |
| 272 | // setReplication records host's replication class, same narrow-UPDATE contract |
| 273 | // as setHibernate. Validated here (not just at the CLI) so no LOCAL write path |
| 274 | // — operator, controld, or a future billing push — can typo a class the wake |
| 275 | // path would silently treat as stream. upsert deliberately does NOT validate: |
| 276 | // a mesh peer on a newer build may replicate a class this build doesn't know, |
| 277 | // and storing it verbatim (replicationSyncInterval fail-safes unknowns to |
| 278 | // stream) keeps the space's failover working through the skew, converging when |
| 279 | // this box updates. An unknown host is a quiet no-op. |
| 280 | // setCanary flips host's throwaway-instance marker — a narrow UPDATE like |
| 281 | // setHibernate, no other field touched. An unknown host is a quiet no-op. |
| 282 | func (r *Registry) setCanary(host string, on bool) error { |
| 283 | _, err := r.sql.Exec(`UPDATE routes SET canary = ? WHERE host = ?`, on, host) |
| 284 | return err |
| 285 | } |
| 286 | |
| 287 | func (r *Registry) setReplication(host, class string) error { |
| 288 | if !validReplication(class) { |
| 289 | return fmt.Errorf("unknown replication class %q (want %s, %s or %s)", |
| 290 | class, replicationStream, replicationBatched, replicationDaily) |
| 291 | } |
| 292 | _, err := r.sql.Exec(`UPDATE routes SET replication = ? WHERE host = ?`, class, host) |
| 293 | return err |
| 294 | } |
| 295 | |
| 296 | // setHomeRegion records host's placement region (multi-edge, docs/multi-edge.md) |
| 297 | // without touching the rest of the route — the same narrow-UPDATE contract as |
| 298 | // setHibernate/setReplication, so `lchat relocate` can re-home a space onto a |
| 299 | // new region without clobbering a concurrent next_due write or a deploy's |
| 300 | // read-modify-write (and vice versa). The mesh reads this replicated column to |
| 301 | // place the space on a box in the matching region (mesh.go's candidatesFor), so |
| 302 | // flipping it IS the whole mechanic of relocation — the hash re-homes the space |
| 303 | // and the single-writer lease still guarantees one writer through the move. An |
| 304 | // unknown host is a quiet no-op, matching setHibernate; region "" means |
| 305 | // region-agnostic (today's default, hashed across the whole fleet). |
| 306 | func (r *Registry) setHomeRegion(host, region string) error { |
| 307 | _, err := r.sql.Exec(`UPDATE routes SET home_region = ? WHERE host = ?`, region, host) |
| 308 | return err |
| 309 | } |
| 310 | |
| 311 | // setVersions records a host's current + previous version, same narrow-UPDATE |
| 312 | // contract as setHibernate/setNextDue. Deploys and update passes hold their |
| 313 | // route reads for a while (minutes, when a download runs) — a full upsert |
| 314 | // would write back next_due/hibernate as they were at READ time, clobbering |
| 315 | // the activator's mid-pass setNextDue (a silently missed scheduled wake) or |
| 316 | // an operator's hibernate flip. |
| 317 | func (r *Registry) setVersions(host, ver, prev string) error { |
| 318 | _, err := r.sql.Exec(`UPDATE routes SET version = ?, prev_version = ? WHERE host = ?`, ver, prev, host) |
| 319 | return err |
| 320 | } |
| 321 | |
| 322 | // ---- due-time index (#550 Part 4 v2 follow-up) -------------------------------- |
| 323 | // |
| 324 | // The activator's persistence for scheduled deliveries on hibernating hosts: |
| 325 | // setNextDue records at sleep time, dueInstances answers the sweeper's "who |
| 326 | // needs waking". Together they implement actDueStore (activator.go). |
| 327 | |
| 328 | // setNextDue records host's next scheduled-delivery time (unix seconds; 0 = |
| 329 | // nothing scheduled) without touching the rest of the route. A host not in the |
| 330 | // registry is a quiet no-op — it can't be woken anyway. |
| 331 | func (r *Registry) setNextDue(host string, due int64) error { |
| 332 | _, err := r.sql.Exec(`UPDATE routes SET next_due = ? WHERE host = ?`, due, host) |
| 333 | return err |
| 334 | } |
| 335 | |
| 336 | // dueInstances returns the ready instance routes whose recorded next_due has |
| 337 | // arrived (0 < next_due <= now) — the sweeper's due-time wake set. Reading the |
| 338 | // registry (not activator memory) each sweep is what makes the index survive |
| 339 | // router restarts and track route removals for free; the fleet is tens of |
| 340 | // routes, so a filtered list per tick is nothing. |
| 341 | func (r *Registry) dueInstances(now int64) ([]Route, error) { |
| 342 | routes, err := r.list() |
| 343 | if err != nil { |
| 344 | return nil, err |
| 345 | } |
| 346 | var due []Route |
| 347 | for _, t := range routes { |
| 348 | // Only hibernating hosts: the index exists to wake sleeping instances, |
| 349 | // and a systemd-managed host is always on — spawning a child for it |
| 350 | // would fight its unit over the socket. This also makes a leftover |
| 351 | // next_due inert on a host flipped back to systemd (the flip-off's |
| 352 | // release re-records one after the flag is already clear). |
| 353 | if t.Kind == "instance" && t.Status == "ready" && t.Hibernate && t.NextDue > 0 && t.NextDue <= now { |
| 354 | due = append(due, t) |
| 355 | } |
| 356 | } |
| 357 | return due, nil |
| 358 | } |
| 359 | |
amadan