| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "net/http" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strconv" |
| 13 | "strings" |
| 14 | "time" |
| 15 | ) |
| 16 | |
| 17 | // Reclaiming never-claimed spaces. |
| 18 | // |
| 19 | // Self-serve creation hands out a slot against the capacity cap the moment a |
| 20 | // space is made. A space whose admin never completes setup — the setup invite |
| 21 | // expires after 7 days — would otherwise sit there forever, holding a slot and a |
| 22 | // disk. The reaper sweeps the cloud ledger for spaces older than a threshold, |
| 23 | // asks the control daemon whether each was ever set up (see instanceStatus — the |
| 24 | // signal is `setup_complete`, NOT member count), and deprovisions the ones that |
| 25 | // never were. Deprovisioning ARCHIVES to /data/trash (deleteInstance never |
| 26 | // destroys), so a false positive is recoverable; a companion trash purge |
| 27 | // reclaims that archive after a longer window. |
| 28 | // |
| 29 | // Both are opt-in (off unless their env is set) and only ever touch spaces IN |
| 30 | // the cloud ledger — operator-run instances (family.…, home, push, …) are never |
| 31 | // in it, so they can never be candidates. |
| 32 | |
| 33 | // ---- "never claimed" signal (controld-side, root) ----------------------------- |
| 34 | |
| 35 | // instanceStatus reads a space's claimed-ness from its DB, READ-ONLY (safe |
| 36 | // against the live instance — no migrations, no writes). For a hibernating |
| 37 | // space this is also wake-free by construction: an asleep instance's local DB |
| 38 | // is the warm cache the activator left at its last sleep (on one box it is the |
| 39 | // newest truth — S3 only ever holds what was checkpointed FROM it), so the |
| 40 | // answer comes straight off the file with no socket touched. A hibernating |
| 41 | // space that has NEVER woken has no DB at all — os.Stat propagates not-exist |
| 42 | // and the stats handler (controld.go) answers definitionally from the registry |
| 43 | // instead: members 0, never set up, no content. |
| 44 | // |
| 45 | // "Never claimed" is NOT "zero members": add-instance pre-seeds the admin user |
| 46 | // so the setup invite has an identity to ship, and claiming BECOMES that same |
| 47 | // user — so a space has exactly one member whether or not anyone ever set it up. |
| 48 | // (The pre-seed happens at setup-invite mint time, i.e. against a RUNNING |
| 49 | // instance: for a hibernating create that's the create-path's readiness probe — |
| 50 | // its first wake — so a woken-at-create space has the one admin; a never-woken |
| 51 | // one has none.) The real signal is `setup_complete`: absent until the admin |
| 52 | // finishes the setup conversation (markSetupComplete), then "1". members is |
| 53 | // returned too, as a belt-and-suspenders guard (don't reap if anyone beyond the |
| 54 | // seeded admin joined). |
| 55 | func instanceStatus(host string) (members, people int, setupComplete, hasContent bool, err error) { |
| 56 | path := filepath.Join(instanceDataDir, host+".db") |
| 57 | if _, err := os.Stat(path); err != nil { |
| 58 | return 0, 0, false, false, err // not-exist propagates → the caller answers unknown_host |
| 59 | } |
| 60 | conn, err := sql.Open("sqlite", "file:"+path+"?mode=ro&_pragma=busy_timeout(3000)") |
| 61 | if err != nil { |
| 62 | return 0, 0, false, false, err |
| 63 | } |
| 64 | defer conn.Close() |
| 65 | conn.SetMaxOpenConns(1) |
| 66 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 67 | defer cancel() |
| 68 | if err := conn.QueryRowContext(ctx, "SELECT COUNT(*) FROM users").Scan(&members); err != nil { |
| 69 | return 0, 0, false, false, err |
| 70 | } |
| 71 | // people is the billable head count: members who aren't agents — per-seat |
| 72 | // billing must never charge for Claude or Eleven. Like the tombstone note |
| 73 | // below, this reads OTHER instances' DB files, which may predate is_agent; |
| 74 | // a failed count falls back to the full member count rather than erroring |
| 75 | // the whole stats read. |
| 76 | people = members |
| 77 | var np int |
| 78 | if err := conn.QueryRowContext(ctx, "SELECT COUNT(*) FROM users WHERE COALESCE(is_agent, 0) = 0").Scan(&np); err == nil { |
| 79 | people = np |
| 80 | } |
| 81 | var v string |
| 82 | switch err := conn.QueryRowContext(ctx, "SELECT value FROM config WHERE key='setup_complete'").Scan(&v); err { |
| 83 | case nil, sql.ErrNoRows: // absent = never set up |
| 84 | default: |
| 85 | return 0, 0, false, false, err |
| 86 | } |
| 87 | // hasContent guards against a false positive: any message or thread means the |
| 88 | // space was actually used, so it's never reaped no matter what setup_complete |
| 89 | // says. EXISTS short-circuits, so this stays O(1) even on a large DB. |
| 90 | // NB: delete tombstones (#581) count as "content" here. That's deliberate: |
| 91 | // this runs against OTHER instances' DB files, which may predate the |
| 92 | // tombstone_of column (filtering on it would error), and a tombstone means |
| 93 | // someone really used the space — erring protective is the guard's job. |
| 94 | var hasMsg, hasThread int |
| 95 | if err := conn.QueryRowContext(ctx, |
| 96 | "SELECT EXISTS(SELECT 1 FROM messages), EXISTS(SELECT 1 FROM threads)").Scan(&hasMsg, &hasThread); err != nil { |
| 97 | return 0, 0, false, false, err |
| 98 | } |
| 99 | return members, people, v == "1", hasMsg == 1 || hasThread == 1, nil |
| 100 | } |
| 101 | |
| 102 | // ---- trash purge (controld-side, root) ---------------------------------------- |
| 103 | |
| 104 | // trashDirTime returns when a /data/trash entry was archived. deleteInstance |
| 105 | // names them "<host>-<unix>"; trust that suffix, falling back to the dir mtime. |
| 106 | func trashDirTime(name string, info os.FileInfo) time.Time { |
| 107 | if i := strings.LastIndexByte(name, '-'); i >= 0 { |
| 108 | if sec, err := strconv.ParseInt(name[i+1:], 10, 64); err == nil && sec > 0 { |
| 109 | return time.Unix(sec, 0) |
| 110 | } |
| 111 | } |
| 112 | if info != nil { |
| 113 | return info.ModTime() |
| 114 | } |
| 115 | return time.Time{} |
| 116 | } |
| 117 | |
| 118 | // purgeTrash removes archived spaces older than keep, returning how many it |
| 119 | // removed. Pure but for the filesystem, so it's directly testable. |
| 120 | func purgeTrash(trashDir string, keep time.Duration, now time.Time) (int, error) { |
| 121 | entries, err := os.ReadDir(trashDir) |
| 122 | if os.IsNotExist(err) { |
| 123 | return 0, nil |
| 124 | } |
| 125 | if err != nil { |
| 126 | return 0, err |
| 127 | } |
| 128 | cutoff := now.Add(-keep) |
| 129 | removed := 0 |
| 130 | for _, e := range entries { |
| 131 | if !e.IsDir() { |
| 132 | continue |
| 133 | } |
| 134 | info, _ := e.Info() |
| 135 | if trashDirTime(e.Name(), info).Before(cutoff) { |
| 136 | if err := os.RemoveAll(filepath.Join(trashDir, e.Name())); err != nil { |
| 137 | log.Printf("controld: trash purge %s: %v", e.Name(), err) |
| 138 | continue |
| 139 | } |
| 140 | removed++ |
| 141 | } |
| 142 | } |
| 143 | return removed, nil |
| 144 | } |
| 145 | |
| 146 | // runTrashPurge sweeps /data/trash once at startup then daily. keep<=0 disables it. |
| 147 | func runTrashPurge(keep time.Duration) { |
| 148 | if keep <= 0 { |
| 149 | return |
| 150 | } |
| 151 | trashDir := filepath.Join(instanceDataDir, "trash") |
| 152 | log.Printf("controld: trash purge on — reclaims archived spaces older than %s", keep) |
| 153 | for { |
| 154 | if n, err := purgeTrash(trashDir, keep, time.Now()); err != nil { |
| 155 | log.Printf("controld: trash purge: %v", err) |
| 156 | } else if n > 0 { |
| 157 | log.Printf("controld: trash purge removed %d archived space(s)", n) |
| 158 | } |
| 159 | time.Sleep(24 * time.Hour) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | // ---- the reaper (cloud-side) --------------------------------------------------- |
| 164 | |
| 165 | // instanceStats is what the control daemon reports about one space: the |
| 166 | // reaper's guards plus the storage meter's byte count (meter.go). |
| 167 | type instanceStats struct { |
| 168 | Members int `json:"members"` |
| 169 | // People is the non-agent member count — what per-seat billing charges |
| 170 | // for (bots are free). Absent from pre-people controld responses, where |
| 171 | // it decodes as 0; consumers fall back to Members. |
| 172 | People int `json:"people"` |
| 173 | SetupComplete bool `json:"setup_complete"` |
| 174 | HasContent bool `json:"has_content"` |
| 175 | StorageBytes int64 `json:"storage_bytes"` |
| 176 | } |
| 177 | |
| 178 | // controlStats asks the control daemon for a space's stats. |
| 179 | func (s *cloudServer) controlStats(host string) (instanceStats, error) { |
| 180 | var st instanceStats |
| 181 | req, err := http.NewRequest("GET", "http://control/instances/"+host+"/stats", nil) |
| 182 | if err != nil { |
| 183 | return st, err |
| 184 | } |
| 185 | req.Header.Set("Authorization", "Bearer "+s.controlToken) |
| 186 | resp, err := s.control.Do(req) |
| 187 | if err != nil { |
| 188 | return st, err |
| 189 | } |
| 190 | defer resp.Body.Close() |
| 191 | if resp.StatusCode != http.StatusOK { |
| 192 | return st, fmt.Errorf("control stats %s: HTTP %d", host, resp.StatusCode) |
| 193 | } |
| 194 | if err := json.NewDecoder(resp.Body).Decode(&st); err != nil { |
| 195 | return st, err |
| 196 | } |
| 197 | return st, nil |
| 198 | } |
| 199 | |
| 200 | // reapLoop sweeps every interval; enabled only when reapAfter>0. |
| 201 | func (s *cloudServer) reapLoop(interval time.Duration) { |
| 202 | if s.reapAfter <= 0 { |
| 203 | return |
| 204 | } |
| 205 | mode := "" |
| 206 | if s.reapDryRun { |
| 207 | mode = " (dry-run — logs only, deletes nothing)" |
| 208 | } |
| 209 | log.Printf("cloud: reaper on%s — reclaims never-claimed spaces older than %s, sweeping every %s", mode, s.reapAfter, interval) |
| 210 | for { |
| 211 | s.reapOnce(time.Now()) |
| 212 | time.Sleep(interval) |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | // reapOnce reclaims never-claimed spaces older than reapAfter. It deletes only a |
| 217 | // space the control daemon confirms was NEVER set up (setup_complete absent), that |
| 218 | // nobody beyond the pre-seeded admin joined (members <= 1), AND that has zero |
| 219 | // content (no messages or threads) — three independent guards, so a space with any |
| 220 | // real use survives all of them. Anything set up / in use / with content / whose |
| 221 | // status can't be read is left untouched (a stats error is a skip, never a reap). |
| 222 | // A per-sweep cap (reapMax) bounds the blast radius of any future mis-classification |
| 223 | // to a handful of recoverable archives before an operator would notice. Returns counts. |
| 224 | func (s *cloudServer) reapOnce(now time.Time) (reaped, kept int) { |
| 225 | cutoff := now.Add(-s.reapAfter).Unix() |
| 226 | cands, err := s.db.spacesOlderThan(cutoff) |
| 227 | if err != nil { |
| 228 | log.Printf("cloud: reaper: ledger query: %v", err) |
| 229 | return 0, 0 |
| 230 | } |
| 231 | for _, c := range cands { |
| 232 | st, err := s.controlStats(c.Host) |
| 233 | if err != nil { |
| 234 | log.Printf("cloud: reaper: skip %s (stats unavailable): %v", c.Host, err) |
| 235 | continue |
| 236 | } |
| 237 | if st.SetupComplete || st.Members > 1 || st.HasContent { |
| 238 | kept++ |
| 239 | continue // set up, someone else joined, or has content — never reap |
| 240 | } |
| 241 | created := time.Unix(c.CreatedAt, 0).UTC().Format(time.RFC3339) |
| 242 | if s.reapDryRun { |
| 243 | log.Printf("cloud: reaper: WOULD reclaim %s (id %d, never set up, created %s)", c.Host, c.ID, created) |
| 244 | continue |
| 245 | } |
| 246 | if s.reapMax > 0 && reaped >= s.reapMax { |
| 247 | log.Printf("cloud: reaper: per-sweep cap (%d) reached — deferring the rest to the next sweep", s.reapMax) |
| 248 | break |
| 249 | } |
| 250 | status, body, err := s.controlDelete(c.Host, false) |
| 251 | if err != nil { |
| 252 | log.Printf("cloud: reaper: delete %s failed: %v", c.Host, err) |
| 253 | continue |
| 254 | } |
| 255 | // unknown_host = already gone (an operator removed it); still clear the row. |
| 256 | if status != http.StatusOK && body["error"] != "unknown_host" { |
| 257 | log.Printf("cloud: reaper: delete %s: HTTP %d", c.Host, status) |
| 258 | continue |
| 259 | } |
| 260 | if err := s.db.deleteSpaceByID(c.ID); err != nil { |
| 261 | log.Printf("cloud: reaper: ledger delete %s (id %d): %v", c.Host, c.ID, err) |
| 262 | } |
| 263 | log.Printf("cloud: reaper: reclaimed never-claimed space %s (id %d, created %s)", c.Host, c.ID, created) |
| 264 | reaped++ |
| 265 | } |
| 266 | if reaped > 0 || kept > 0 { |
| 267 | log.Printf("cloud: reaper: swept %d candidate(s) — %d reclaimed, %d still in use", len(cands), reaped, kept) |
| 268 | } |
| 269 | return reaped, kept |
| 270 | } |
| 271 | |
| 272 | // ---- self-reclaim (the create path's escape hatch) ------------------------------ |
| 273 | |
| 274 | // reclaimAge is how old one of the account's own spaces must be before a |
| 275 | // quota-blocked create may reclaim it. Setup takes minutes, so an hour cleanly |
| 276 | // separates "abandoned yesterday, retrying now" (reclaim) from "setting it up |
| 277 | // in another tab right now" (never touch) — and the home page keeps the setup |
| 278 | // link for continuing inside the window. |
| 279 | const reclaimAge = time.Hour |
| 280 | |
| 281 | // reclaimOwnSpace frees at most one quota slot for (caller, requester) by |
| 282 | // reclaiming their own oldest abandoned allocation: a pending row whose create |
| 283 | // died between allocate and finalize, or a provisioned space that clears the |
| 284 | // reaper's never-used guards (never set up, nobody joined, zero content) — |
| 285 | // plus ownership and the age gate on top. Data still archives to /data/trash |
| 286 | // via controld, same as the reaper. A stats error is a skip, never a reclaim. |
| 287 | // Returns whether a slot was freed. |
| 288 | func (s *cloudServer) reclaimOwnSpace(callerID, requester string) bool { |
| 289 | cands, err := s.db.reclaimCandidates(callerID, requester, time.Now().Add(-reclaimAge).Unix()) |
| 290 | if err != nil { |
| 291 | log.Printf("cloud: self-reclaim: ledger query: %v", err) |
| 292 | return false |
| 293 | } |
| 294 | for _, c := range cands { |
| 295 | if c.Host == "" { |
| 296 | // A pending row this old is a dead create (real ones finalize in |
| 297 | // seconds); there is no instance behind it — ledger-only cleanup. |
| 298 | if err := s.db.deleteSpaceByID(c.ID); err != nil { |
| 299 | log.Printf("cloud: self-reclaim: clear stale pending row (id %d): %v", c.ID, err) |
| 300 | continue |
| 301 | } |
| 302 | log.Printf("cloud: self-reclaim: cleared stale pending allocation (id %d, requester of caller %.8s…)", c.ID, callerID) |
| 303 | return true |
| 304 | } |
| 305 | st, err := s.controlStats(c.Host) |
| 306 | if err != nil { |
| 307 | log.Printf("cloud: self-reclaim: skip %s (stats unavailable): %v", c.Host, err) |
| 308 | continue |
| 309 | } |
| 310 | if st.SetupComplete || st.Members > 1 || st.HasContent { |
| 311 | continue // in real use — the quota refusal stands |
| 312 | } |
| 313 | status, body, err := s.controlDelete(c.Host, false) |
| 314 | if err != nil { |
| 315 | log.Printf("cloud: self-reclaim: delete %s failed: %v", c.Host, err) |
| 316 | continue |
| 317 | } |
| 318 | if status != http.StatusOK && body["error"] != "unknown_host" { |
| 319 | log.Printf("cloud: self-reclaim: delete %s: HTTP %d", c.Host, status) |
| 320 | continue |
| 321 | } |
| 322 | if err := s.db.deleteSpaceByID(c.ID); err != nil { |
| 323 | log.Printf("cloud: self-reclaim: ledger delete %s (id %d): %v", c.Host, c.ID, err) |
| 324 | } |
| 325 | log.Printf("cloud: self-reclaim: reclaimed %s (id %d) — owner's own never-set-up space, making room for their new create", c.Host, c.ID) |
| 326 | return true |
| 327 | } |
| 328 | return false |
| 329 | } |
| 330 | |
amadan