| 1 | // 🤖 Package activator holds the replication primitives for instance |
| 2 | // hibernation (issue #550 Part 4 v2): litestream embedded as a library — pure |
| 3 | // modernc.org/sqlite, no cgo — with its S3 lease (lock.json via conditional |
| 4 | // writes) wrapped around replication as a single-writer lock. The lease is |
| 5 | // the truth of ownership: a second claimant fails loudly with a typed |
| 6 | // *litestream.LeaseExistsError (the CLI, by contrast, silently absorbs a |
| 7 | // second replicator — proven on #550), and a failed renewal means leadership |
| 8 | // is lost and the caller must stand down. The Phase 2 activator (wake/sleep, |
| 9 | // activator.go in the main package) builds on these primitives. |
| 10 | package activator |
| 11 | |
| 12 | import ( |
| 13 | "context" |
| 14 | "errors" |
| 15 | "fmt" |
| 16 | "path/filepath" |
| 17 | "sync" |
| 18 | "time" |
| 19 | |
| 20 | awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| 21 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 22 | "github.com/benbjohnson/litestream" |
| 23 | lss3 "github.com/benbjohnson/litestream/s3" |
| 24 | ) |
| 25 | |
| 26 | // LeaseTTL is deliberately long — 30× litestream's 30s default. Renewal runs |
| 27 | // at TTL/3 and each renewal is one S3 Tier-1 PUT, so at the old 30s TTL the |
| 28 | // lease heartbeat alone was 8,640 PUTs/day per awake instance (~$16/yr — |
| 29 | // ~70% of the fleet's entire S3 bill, and more than a €5/y space's revenue); |
| 30 | // at 15m it's 288/day (~$0.50/yr). The price is the failover window: a |
| 31 | // HARD-crashed holder (kill -9, box death) strands the lease until expiry, |
| 32 | // so a peer's takeover — or a re-wake on the same box — can stall up to |
| 33 | // ~15m. Every graceful path (sleep, deploy, shutdown) releases the lease |
| 34 | // explicitly and is unaffected. Shorten this only if fast crash-failover |
| 35 | // ever outranks per-instance cost. |
| 36 | const LeaseTTL = 15 * time.Minute |
| 37 | |
| 38 | // compactionLevels replaces litestream.DefaultCompactionLevels (L1 30s / |
| 39 | // L2 5m / L3 1h) with a far gentler ladder. The defaults are tuned for a |
| 40 | // long-lived replicator where background compaction promptness matters; a |
| 41 | // hibernating instance instead checkpoints deterministically at sleep via |
| 42 | // SyncAndWait (which calls db.Sync + Replica.Sync directly, independent of |
| 43 | // every interval here), so aggressive background compaction buys nothing — |
| 44 | // it only burns S3 Tier-1 LISTs: each level's monitor tick costs up to two |
| 45 | // LISTs per DB even with zero writes — L1 at 30s dominated the fleet's idle |
| 46 | // request bill, and the first fix (5m/30m/3h) still left ~1.3k LISTs/day per |
| 47 | // awake instance once the lease PUTs were tamed. |
| 48 | // Correctness is untouched — compaction is purely a restore-time |
| 49 | // read-amplification optimisation (restore replays snapshot + whatever L0 |
| 50 | // files exist), and each level's monitor still runs one immediate pass at |
| 51 | // store.Open, so every wake compacts anything a previous short awake window |
| 52 | // left pending. Same three-level geometric shape, so restore reads stay |
| 53 | // bounded: worst case a hard crash replays up to ~an hour of L0 files. |
| 54 | var compactionLevels = litestream.CompactionLevels{ |
| 55 | {Level: 0, Interval: 0}, |
| 56 | {Level: 1, Interval: 1 * time.Hour}, |
| 57 | {Level: 2, Interval: 6 * time.Hour}, |
| 58 | {Level: 3, Interval: 24 * time.Hour}, |
| 59 | } |
| 60 | |
| 61 | // dbMonitorInterval slows litestream's DB monitor (default 1s) — the local |
| 62 | // WAL poll that shovels new frames into shadow LTX. It makes no S3 calls, |
| 63 | // but 1s polling per awake instance is pointless churn when the sleep-time |
| 64 | // SyncAndWait ships everything regardless; 10s bounds an awake instance's |
| 65 | // crash-loss window to ~10s, plenty tight for chat. |
| 66 | // (Replica.SyncInterval stays at its 1s default on purpose: idle replica |
| 67 | // ticks make zero S3 calls — the upload loop no-ops when the local TXID |
| 68 | // hasn't moved — and while awake it bounds crash-loss to ~1s.) |
| 69 | const dbMonitorInterval = 10 * time.Second |
| 70 | |
| 71 | // l0RetentionCheckInterval slows the store's L0 retention monitor (default |
| 72 | // 15s — tuned for litestream's VFS read replicas, which we don't run). Every |
| 73 | // tick LISTs L1 and usually L0 per DB even with zero writes — the single |
| 74 | // biggest idle-request source. L0 files only become reclaimable after an L1 |
| 75 | // compaction, so checking more often than the L1 interval (1h above) is pure |
| 76 | // waste; matching it cuts the LISTs 240× vs the default with no change to |
| 77 | // what gets deleted. |
| 78 | const l0RetentionCheckInterval = 1 * time.Hour |
| 79 | |
| 80 | // renewer keeps a held lease alive on a ticker. A failed renewal is recorded |
| 81 | // and renewal stops — leadership is lost and the owner must stand down; |
| 82 | // retrying could fence with a new claimant who took over after expiry. |
| 83 | type renewer struct { |
| 84 | leaser *lss3.Leaser |
| 85 | |
| 86 | mu sync.Mutex |
| 87 | lease *litestream.Lease |
| 88 | err error // set if a renewal ever fails (leadership lost) |
| 89 | |
| 90 | cancel context.CancelFunc |
| 91 | done chan struct{} |
| 92 | } |
| 93 | |
| 94 | func startRenewer(leaser *lss3.Leaser, lease *litestream.Lease) *renewer { |
| 95 | ctx, cancel := context.WithCancel(context.Background()) |
| 96 | rn := &renewer{leaser: leaser, lease: lease, cancel: cancel, done: make(chan struct{})} |
| 97 | go func() { |
| 98 | defer close(rn.done) |
| 99 | t := time.NewTicker(LeaseTTL / 3) |
| 100 | defer t.Stop() |
| 101 | for { |
| 102 | select { |
| 103 | case <-ctx.Done(): |
| 104 | return |
| 105 | case <-t.C: |
| 106 | rn.mu.Lock() |
| 107 | cur := rn.lease |
| 108 | rn.mu.Unlock() |
| 109 | renewed, err := rn.leaser.RenewLease(ctx, cur) |
| 110 | rn.mu.Lock() |
| 111 | if err != nil { |
| 112 | rn.err = err |
| 113 | } else { |
| 114 | rn.lease = renewed |
| 115 | } |
| 116 | rn.mu.Unlock() |
| 117 | if err != nil { |
| 118 | return |
| 119 | } |
| 120 | } |
| 121 | } |
| 122 | }() |
| 123 | return rn |
| 124 | } |
| 125 | |
| 126 | // Err reports whether a renewal has failed (leadership lost). |
| 127 | func (rn *renewer) Err() error { |
| 128 | rn.mu.Lock() |
| 129 | defer rn.mu.Unlock() |
| 130 | return rn.err |
| 131 | } |
| 132 | |
| 133 | // stop halts renewal and returns the most recent lease (for release, or to |
| 134 | // hand renewal duty to a successor renewer). |
| 135 | func (rn *renewer) stop() *litestream.Lease { |
| 136 | rn.cancel() |
| 137 | <-rn.done |
| 138 | rn.mu.Lock() |
| 139 | defer rn.mu.Unlock() |
| 140 | return rn.lease |
| 141 | } |
| 142 | |
| 143 | // HeldLease is an acquired single-writer lease that is not yet coupled to |
| 144 | // replication. The wake path needs this split: the DB restore must happen |
| 145 | // *under* the lock (restoring without it could later shadow newer S3 state |
| 146 | // written by the true holder), but replication can only start once the DB |
| 147 | // file exists. Renewal runs from acquisition, so a slow restore can't silently |
| 148 | // outlive the TTL. |
| 149 | type HeldLease struct { |
| 150 | leaser *lss3.Leaser |
| 151 | region string // the leaser's client keeps its region private; carry it for Replicate |
| 152 | rn *renewer |
| 153 | } |
| 154 | |
| 155 | // AcquireLease claims the single-writer lease at s3://bucket/prefix/lock.json |
| 156 | // and starts renewing it. If another process holds the lease it fails LOUDLY |
| 157 | // with a *litestream.LeaseExistsError (check with errors.As) — this is the |
| 158 | // signal the activator reacts to. Region may be empty; the AWS default chain |
| 159 | // (env/profile) resolves it. |
| 160 | func AcquireLease(ctx context.Context, bucket, prefix, region string) (*HeldLease, error) { |
| 161 | // The leaser wants a ready-made S3 client (unlike the replica client, |
| 162 | // which builds its own). Default AWS credential chain. |
| 163 | var opts []func(*awsconfig.LoadOptions) error |
| 164 | if region != "" { |
| 165 | opts = append(opts, awsconfig.WithRegion(region)) |
| 166 | } |
| 167 | cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...) |
| 168 | if err != nil { |
| 169 | return nil, fmt.Errorf("load aws config: %w", err) |
| 170 | } |
| 171 | |
| 172 | leaser := lss3.NewLeaser() |
| 173 | leaser.SetClient(awss3.NewFromConfig(cfg)) |
| 174 | leaser.Bucket = bucket |
| 175 | leaser.Path = prefix // lock lives at <prefix>/lock.json, colocated with the replica |
| 176 | leaser.TTL = LeaseTTL |
| 177 | |
| 178 | lease, err := leaser.AcquireLease(ctx) |
| 179 | if err != nil { |
| 180 | // Includes *litestream.LeaseExistsError when someone else holds it. |
| 181 | return nil, fmt.Errorf("acquire lease: %w", err) |
| 182 | } |
| 183 | return &HeldLease{leaser: leaser, region: region, rn: startRenewer(leaser, lease)}, nil |
| 184 | } |
| 185 | |
| 186 | // Err reports whether lease renewal has failed (leadership lost). |
| 187 | func (h *HeldLease) Err() error { return h.rn.Err() } |
| 188 | |
| 189 | // Release stops renewal and releases the lease — the bail-out when a wake |
| 190 | // fails between acquisition and replication. |
| 191 | func (h *HeldLease) Release(ctx context.Context) error { |
| 192 | lease := h.rn.stop() |
| 193 | if lease == nil { |
| 194 | return nil |
| 195 | } |
| 196 | return h.leaser.ReleaseLease(ctx, lease) |
| 197 | } |
| 198 | |
| 199 | // Replicate trades the held lease for running replication of dbPath to the |
| 200 | // lease's S3 prefix. On success the HeldLease is spent — renewal duty moves to |
| 201 | // the returned LeasedReplicator (Close it to release). On failure the lease is |
| 202 | // STILL HELD; the caller decides whether to retry or Release. |
| 203 | // |
| 204 | // syncInterval sets litestream's replica sync tick — the per-plan durability |
| 205 | // ladder's one knob. It is a MINIMUM spacing between S3 syncs (the monitor's |
| 206 | // change-notify only short-circuits the idle wait, never this spacing), so 5m |
| 207 | // bounds an awake host's hard-crash loss to ~5m and 24h to ~a day, while the |
| 208 | // wake-time initial pass and the sleep-time SyncAndWait/Close flush in full |
| 209 | // regardless — the interval never weakens hibernation's parked copy. 0 keeps |
| 210 | // litestream's 1s default (the stream class): idle ticks make zero S3 calls, |
| 211 | // so stream's request bill is change-driven, not clock-driven. |
| 212 | func (h *HeldLease) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (*LeasedReplicator, error) { |
| 213 | if err := h.rn.Err(); err != nil { |
| 214 | return nil, fmt.Errorf("lease already lost: %w", err) |
| 215 | } |
| 216 | |
| 217 | db := litestream.NewDB(dbPath) |
| 218 | // The activator's replication state lives in its OWN meta directory — |
| 219 | // litestream derives ".<db>-litestream" beside the DB by default, which any |
| 220 | // other litestream process over the same file (the box-wide backup |
| 221 | // replicator, before or after a handover) also uses. Sharing corrupts LTX |
| 222 | // state even SEQUENTIALLY: the fleet-wide flip left a dozen hosts in an |
| 223 | // error loop until their meta dirs were cleared by hand. A distinct |
| 224 | // namespace ends the class. |
| 225 | db.SetMetaPath(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream")) |
| 226 | db.MonitorInterval = dbMonitorInterval |
| 227 | client := lss3.NewReplicaClient() |
| 228 | client.Bucket = h.leaser.Bucket |
| 229 | client.Path = h.leaser.Path |
| 230 | client.Region = h.region |
| 231 | db.Replica = litestream.NewReplicaWithClient(db, client) |
| 232 | if syncInterval > 0 { |
| 233 | db.Replica.SyncInterval = syncInterval |
| 234 | } |
| 235 | |
| 236 | store := litestream.NewStore([]*litestream.DB{db}, compactionLevels) |
| 237 | store.L0RetentionCheckInterval = l0RetentionCheckInterval |
| 238 | if err := store.Open(ctx); err != nil { |
| 239 | return nil, fmt.Errorf("open litestream store: %w", err) |
| 240 | } |
| 241 | |
| 242 | // Hand renewal to the replicator. The handoff gap between stopping one |
| 243 | // renewer and starting the next is microseconds against a 30s TTL. |
| 244 | lease := h.rn.stop() |
| 245 | return &LeasedReplicator{leaser: h.leaser, store: store, db: db, rn: startRenewer(h.leaser, lease)}, nil |
| 246 | } |
| 247 | |
| 248 | // LeasedReplicator couples an exclusive S3 lease with a running litestream |
| 249 | // replication of one SQLite database to the same S3 prefix. The lease is |
| 250 | // acquired *before* replication starts and renewed on a ticker; Close stops |
| 251 | // replication and releases the lease. |
| 252 | type LeasedReplicator struct { |
| 253 | leaser *lss3.Leaser |
| 254 | store *litestream.Store |
| 255 | db *litestream.DB |
| 256 | rn *renewer |
| 257 | } |
| 258 | |
| 259 | // ReplicateWithLease acquires the single-writer lease at |
| 260 | // s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream |
| 261 | // replication of dbPath to s3://bucket/prefix at the default (stream-class) |
| 262 | // sync interval. If another process holds the lease it fails LOUDLY with a |
| 263 | // *litestream.LeaseExistsError (check with errors.As) — this is the signal |
| 264 | // the activator reacts to. |
| 265 | func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*LeasedReplicator, error) { |
| 266 | held, err := AcquireLease(ctx, bucket, prefix, region) |
| 267 | if err != nil { |
| 268 | return nil, err |
| 269 | } |
| 270 | r, err := held.Replicate(ctx, dbPath, 0) |
| 271 | if err != nil { |
| 272 | _ = held.Release(ctx) |
| 273 | return nil, err |
| 274 | } |
| 275 | return r, nil |
| 276 | } |
| 277 | |
| 278 | // SyncAndWait forces a full sync of the database to the replica and blocks |
| 279 | // until the remote position catches up — used by tests to make replication |
| 280 | // observable without waiting on monitor intervals. |
| 281 | func (r *LeasedReplicator) SyncAndWait(ctx context.Context) error { |
| 282 | return r.db.SyncAndWait(ctx) |
| 283 | } |
| 284 | |
| 285 | // Err reports whether lease renewal has failed (leadership lost). |
| 286 | func (r *LeasedReplicator) Err() error { return r.rn.Err() } |
| 287 | |
| 288 | // Abandon stops lease renewal and replication WITHOUT the final sync or the |
| 289 | // lease release — the wreckage a crashed box leaves behind: lock.json simply |
| 290 | // expires on its TTL, and the replica holds whatever was last synced. It |
| 291 | // exists so tests can simulate hard box death in-process; production code has |
| 292 | // no reason to call it (a real crash needs no help). |
| 293 | func (r *LeasedReplicator) Abandon(ctx context.Context) { |
| 294 | r.rn.stop() // deliberately no ReleaseLease — the TTL is the point |
| 295 | // litestream's only teardown is Close, which includes a sync a real crash |
| 296 | // would skip; it ships nothing as long as the caller killed the writer |
| 297 | // first, and reclaims the replication goroutines either way. |
| 298 | _ = r.store.Close(ctx) |
| 299 | } |
| 300 | |
| 301 | // Close stops replication (final sync included via litestream's DB.Close) |
| 302 | // and releases the lease so the next claimant can acquire immediately. |
| 303 | func (r *LeasedReplicator) Close(ctx context.Context) error { |
| 304 | // Force one full sync before teardown. litestream's DB.Close only |
| 305 | // final-syncs an *initialized* DB, and initialization happens on the |
| 306 | // monitor's first pass — so a checkpoint within ~a second of Replicate |
| 307 | // could silently ship nothing (a drain right after a wake, caught by the |
| 308 | // mesh live test: the replica held no restorable snapshot). SyncAndWait |
| 309 | // forces init + a replica sync; its error is deliberately dropped — with |
| 310 | // the DB now initialized, store.Close's own final sync is the one that |
| 311 | // counts, and its error is returned. |
| 312 | _ = r.db.SyncAndWait(ctx) |
| 313 | err := r.store.Close(ctx) |
| 314 | |
| 315 | lease := r.rn.stop() |
| 316 | if lease != nil { |
| 317 | // The release gets its OWN budget, never the caller's: a wedged final |
| 318 | // sync exhausts ctx, and releasing under a dead context leaves the |
| 319 | // lease held until TTL — the next wake then 503s for up to the full |
| 320 | // TTL over nothing (hit live on the first real instance flip). The |
| 321 | // release is |
| 322 | // one small S3 DELETE; five seconds is generous. |
| 323 | relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 324 | defer cancel() |
| 325 | if e := r.leaser.ReleaseLease(relCtx, lease); e != nil && err == nil { |
| 326 | err = e |
| 327 | } |
| 328 | } |
| 329 | return err |
| 330 | } |
| 331 | |
| 332 | // Restore pulls the newest replicated state of s3://bucket/prefix into |
| 333 | // dbPath via litestream's restore API — the cold-wake path when the local DB |
| 334 | // is missing. Returns restored=false with a nil error when the prefix holds |
| 335 | // no backup at all (a brand-new instance that has never replicated boots |
| 336 | // fresh instead). The caller must hold the prefix's lease: restoring outside |
| 337 | // it could resurrect a stale copy that later shadows the true holder's newer |
| 338 | // state. |
| 339 | func Restore(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error) { |
| 340 | client := lss3.NewReplicaClient() |
| 341 | client.Bucket = bucket |
| 342 | client.Path = prefix |
| 343 | client.Region = region |
| 344 | |
| 345 | // A nil *DB is litestream's own pattern for URL restores (the CLI does the |
| 346 | // same): Restore only needs the replica client and the output path. |
| 347 | r := litestream.NewReplicaWithClient(nil, client) |
| 348 | opt := litestream.NewRestoreOptions() |
| 349 | opt.OutputPath = dbPath |
| 350 | if err := r.Restore(ctx, opt); err != nil { |
| 351 | if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) { |
| 352 | return false, nil |
| 353 | } |
| 354 | return false, err |
| 355 | } |
| 356 | return true, nil |
| 357 | } |
| 358 | |
amadan