| 1 | // Ported from lchat internal/activator/replica.go. |
| 2 | // |
| 3 | // Package lease holds the replication primitives for instance |
| 4 | // hibernation (issue #550 Part 4 v2): litestream embedded as a library — pure |
| 5 | // modernc.org/sqlite, no cgo — with its S3 lease (lock.json via conditional |
| 6 | // writes) wrapped around replication as a single-writer lock. The lease is |
| 7 | // the truth of ownership: a second claimant fails loudly with a typed |
| 8 | // *HeldError (the CLI, by contrast, silently absorbs a |
| 9 | // second replicator — proven on #550), and a failed renewal means leadership |
| 10 | // is lost and the caller must stand down. The activator (wake/sleep) |
| 11 | // builds on these primitives. |
| 12 | package lease |
| 13 | |
| 14 | import ( |
| 15 | "context" |
| 16 | "errors" |
| 17 | "fmt" |
| 18 | "log" |
| 19 | "os" |
| 20 | "path/filepath" |
| 21 | "sync" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/benbjohnson/litestream" |
| 25 | lss3 "github.com/benbjohnson/litestream/s3" |
| 26 | ) |
| 27 | |
| 28 | // LeaseTTL is deliberately long — 30× litestream's 30s default. Renewal runs |
| 29 | // at TTL/3 and each renewal is one S3 Tier-1 PUT, so at the old 30s TTL the |
| 30 | // lease heartbeat alone was 8,640 PUTs/day per awake instance (~$16/yr — |
| 31 | // ~70% of the fleet's entire S3 bill, and more than a €5/y space's revenue); |
| 32 | // at 15m it's 288/day (~$0.50/yr). The price is the failover window: a |
| 33 | // HARD-crashed holder (kill -9, box death) strands the lease until expiry, |
| 34 | // so a peer's takeover — or a re-wake on the same box — can stall up to |
| 35 | // ~15m. Every graceful path (sleep, deploy, shutdown) releases the lease |
| 36 | // explicitly and is unaffected. Shorten this only if fast crash-failover |
| 37 | // ever outranks per-instance cost. |
| 38 | const LeaseTTL = 15 * time.Minute |
| 39 | |
| 40 | // compactionLevels replaces litestream.DefaultCompactionLevels (L1 30s / |
| 41 | // L2 5m / L3 1h) with a far gentler ladder. The defaults are tuned for a |
| 42 | // long-lived replicator where background compaction promptness matters; a |
| 43 | // hibernating instance instead checkpoints deterministically at sleep via |
| 44 | // SyncAndWait (which calls db.Sync + Replica.Sync directly, independent of |
| 45 | // every interval here), so aggressive background compaction buys nothing — |
| 46 | // it only burns S3 Tier-1 LISTs: each level's monitor tick costs up to two |
| 47 | // LISTs per DB even with zero writes — L1 at 30s dominated the fleet's idle |
| 48 | // request bill, and the first fix (5m/30m/3h) still left ~1.3k LISTs/day per |
| 49 | // awake instance once the lease PUTs were tamed. |
| 50 | // Correctness is untouched — compaction is purely a restore-time |
| 51 | // read-amplification optimisation (restore replays snapshot + whatever L0 |
| 52 | // files exist), and each level's monitor still runs one immediate pass at |
| 53 | // store.Open, so every wake compacts anything a previous short awake window |
| 54 | // left pending. Same three-level geometric shape, so restore reads stay |
| 55 | // bounded: worst case a hard crash replays up to ~an hour of L0 files. |
| 56 | var compactionLevels = litestream.CompactionLevels{ |
| 57 | {Level: 0, Interval: 0}, |
| 58 | {Level: 1, Interval: 1 * time.Hour}, |
| 59 | {Level: 2, Interval: 6 * time.Hour}, |
| 60 | {Level: 3, Interval: 24 * time.Hour}, |
| 61 | } |
| 62 | |
| 63 | // SnapshotInterval and SnapshotRetention are the fleet's point-in-time |
| 64 | // recovery window, for every replica the platform writes: this store (a |
| 65 | // hibernating instance's) and the box-wide /etc/litestream.yml, which |
| 66 | // internal/ops writes from these same two constants so they cannot drift. |
| 67 | // |
| 68 | // litestream 0.5's DefaultSnapshotRetention is 24h. That covers a dead box; |
| 69 | // it does not cover LOGICAL corruption, where integrity_check says "ok" and |
| 70 | // the data is simply wrong. A bad migration or an errant DELETE is |
| 71 | // typically noticed days later, and by then the only copy is the |
| 72 | // faithfully-replicated bad one. Here the default was worse than a stated |
| 73 | // day: retention runs in the snapshot monitor, which runs only while the |
| 74 | // instance is awake, so each wake pruned every snapshot older than a day |
| 75 | // (#263), and `carlos db pull --at` reached back about a day. |
| 76 | // |
| 77 | // A week is the deliberate choice. It adds no requests: the snapshot |
| 78 | // monitor's cadence is SnapshotInterval either way, and only how far back |
| 79 | // its deletes reach changes. It does add storage, and more than the seven |
| 80 | // snapshots: retention keeps every L1-L3 file (compactionLevels: hourly, |
| 81 | // 6-hourly, daily) back to the oldest kept snapshot too. Each holds only |
| 82 | // the pages that changed in its window, so a typical small, lightly |
| 83 | // written database adds little. The worst case is a database rewriting |
| 84 | // most of its pages every hour: about 168 L1 + 28 L2 + 7 L3 files plus 7 |
| 85 | // snapshots, some 200 database sizes (100 MiB -> ~20 GiB, ~$0.50/mo at |
| 86 | // ~$0.023/GB-mo), against about 30 under the old 24h. Lengthen it if a |
| 87 | // restore is ever wanted from further back than "we noticed within the |
| 88 | // week"; shorten it if the bucket's size says the worst case is common. |
| 89 | // (Astra's review of #263 corrected an estimate that counted only the |
| 90 | // snapshots.) |
| 91 | const ( |
| 92 | SnapshotInterval = 24 * time.Hour |
| 93 | SnapshotRetention = 7 * 24 * time.Hour |
| 94 | ) |
| 95 | |
| 96 | // dbMonitorInterval slows litestream's DB monitor (default 1s) — the local |
| 97 | // WAL poll that shovels new frames into shadow LTX. It makes no S3 calls, |
| 98 | // but 1s polling per awake instance is pointless churn when the sleep-time |
| 99 | // SyncAndWait ships everything regardless; 10s bounds an awake instance's |
| 100 | // crash-loss window to ~10s, plenty tight for chat. |
| 101 | // (Replica.SyncInterval stays at its 1s default on purpose: idle replica |
| 102 | // ticks make zero S3 calls — the upload loop no-ops when the local TXID |
| 103 | // hasn't moved — and while awake it bounds crash-loss to ~1s.) |
| 104 | const dbMonitorInterval = 10 * time.Second |
| 105 | |
| 106 | // l0RetentionCheckInterval slows the store's L0 retention monitor (default |
| 107 | // 15s — tuned for litestream's VFS read replicas, which we don't run). Every |
| 108 | // tick LISTs L1 and usually L0 per DB even with zero writes — the single |
| 109 | // biggest idle-request source. L0 files only become reclaimable after an L1 |
| 110 | // compaction, so checking more often than the L1 interval (1h above) is pure |
| 111 | // waste; matching it cuts the LISTs 240× vs the default with no change to |
| 112 | // what gets deleted. |
| 113 | const l0RetentionCheckInterval = 1 * time.Hour |
| 114 | |
| 115 | // HeldError reports that another process holds the single-writer lease. It |
| 116 | // wraps litestream's *LeaseExistsError so callers of this package never need |
| 117 | // to import litestream to react to a contested lease (check with errors.As). |
| 118 | type HeldError struct { |
| 119 | Owner string |
| 120 | ExpiresAt time.Time |
| 121 | } |
| 122 | |
| 123 | func (e *HeldError) Error() string { |
| 124 | if e.Owner != "" { |
| 125 | return fmt.Sprintf("lease already held by %s until %s", e.Owner, e.ExpiresAt.Format(time.RFC3339)) |
| 126 | } |
| 127 | return fmt.Sprintf("lease already held until %s", e.ExpiresAt.Format(time.RFC3339)) |
| 128 | } |
| 129 | |
| 130 | // Held is an acquired single-writer lease that is not yet coupled to |
| 131 | // replication. The wake path needs this split: the DB restore must happen |
| 132 | // *under* the lock (restoring without it could later shadow newer S3 state |
| 133 | // written by the true holder), but replication can only start once the DB |
| 134 | // file exists. Renewal runs from acquisition, so a slow restore can't silently |
| 135 | // outlive the TTL. |
| 136 | type Held struct { |
| 137 | leaser *lss3.Leaser |
| 138 | region string // the leaser's client keeps its region private; carry it for Replicate |
| 139 | rn *renewer |
| 140 | mu sync.Mutex |
| 141 | busy bool |
| 142 | spent bool |
| 143 | poison error |
| 144 | replicaClient litestream.ReplicaClient // private local-replica test seam |
| 145 | generationPrefix string |
| 146 | // owed is the lease record a failed Release could not delete (#292). |
| 147 | // Release spends the handle either way; RetryRelease re-issues the |
| 148 | // same conditional delete until it lands or the lease is no longer |
| 149 | // ours. Guarded by mu. |
| 150 | owed *litestream.Lease |
| 151 | } |
| 152 | |
| 153 | // Acquire claims the single-writer lease at s3://bucket/prefix/lock.json |
| 154 | // and starts renewing it. If another process holds the lease it fails LOUDLY |
| 155 | // with a *HeldError (check with errors.As) — this is the |
| 156 | // signal the activator reacts to. Region may be empty; the AWS default chain |
| 157 | // (env/profile) resolves it, or — under a CARLOS_S3_ENDPOINT override, where |
| 158 | // there is no chain to resolve from — endpoint.go's non-AWS default. |
| 159 | func Acquire(ctx context.Context, bucket, prefix, region string) (*Held, error) { |
| 160 | return AcquireTTL(ctx, bucket, prefix, region, LeaseTTL) |
| 161 | } |
| 162 | |
| 163 | // AcquireTTL is Acquire with the lease TTL as a parameter. Production |
| 164 | // Acquire passes LeaseTTL; the moves staging-work lease passes its longer |
| 165 | // term (moves.StagingLeaseTTL, whose renewals have to leave a SigV4 |
| 166 | // signature's whole life in front of SafeUntil); tests pass a short TTL so |
| 167 | // an expiry-and-takeover sequence (the failover race) fits inside a test |
| 168 | // run instead of 15 minutes. The TTL governs renewal spacing too: |
| 169 | // startRenewer ticks at leaser.TTL/3, so a short-TTL lease is renewed |
| 170 | // proportionally often and lapses proportionally fast, which is what |
| 171 | // makes "the holder went dark" observable. |
| 172 | func AcquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Duration) (*Held, error) { |
| 173 | // The leaser wants a ready-made S3 client (unlike the replica client, |
| 174 | // which builds its own). Default AWS credential chain, and the |
| 175 | // CARLOS_S3_ENDPOINT override applied — see endpoint.go for why the lock |
| 176 | // and the data it guards must resolve to the same store. The region is |
| 177 | // resolved ONCE, here, and carried on the Held: Replicate's replica client |
| 178 | // has to sign for whatever this client signed for. |
| 179 | began := time.Now() |
| 180 | client, region, err := newLeaserClient(ctx, region) |
| 181 | if err != nil { |
| 182 | return nil, err |
| 183 | } |
| 184 | built := time.Now() |
| 185 | |
| 186 | leaser := lss3.NewLeaser() |
| 187 | leaser.SetClient(client) |
| 188 | leaser.Bucket = bucket |
| 189 | leaser.Path = prefix // lock lives at <prefix>/lock.json, colocated with the replica |
| 190 | leaser.TTL = ttl |
| 191 | |
| 192 | started := time.Now() |
| 193 | lease, err := createLease(ctx, client, leaser) |
| 194 | if errors.Is(err, errLockPresent) { |
| 195 | lease, err = leaser.AcquireLease(ctx) |
| 196 | } |
| 197 | if err != nil { |
| 198 | // A *litestream.LeaseExistsError means someone else holds it — |
| 199 | // re-surfaced as *HeldError so callers stay litestream-free. |
| 200 | var lexists *litestream.LeaseExistsError |
| 201 | if errors.As(err, &lexists) { |
| 202 | return nil, fmt.Errorf("acquire lease: %w", &HeldError{Owner: lexists.Owner, ExpiresAt: lexists.ExpiresAt}) |
| 203 | } |
| 204 | return nil, fmt.Errorf("acquire lease: %w", err) |
| 205 | } |
| 206 | rn := startRenewer(leaser, lease, started, marginForTTL(ttl)) |
| 207 | if err := rn.Err(); err != nil { |
| 208 | cur := rn.stop() |
| 209 | relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 210 | defer cancel() |
| 211 | _ = leaser.ReleaseLease(relCtx, cur) |
| 212 | return nil, err |
| 213 | } |
| 214 | // Journal only: the wake's lease step, split into the client and the |
| 215 | // lock's own read and conditional write. |
| 216 | log.Printf("lease: %s: acquired in %s (client %s, lock read and write %s)", prefix, |
| 217 | time.Since(began).Round(time.Millisecond), built.Sub(began).Round(time.Millisecond), time.Since(started).Round(time.Millisecond)) |
| 218 | return &Held{leaser: leaser, region: region, rn: rn}, nil |
| 219 | } |
| 220 | |
| 221 | // Err refuses a spent or failed handle as well as lost ownership. |
| 222 | func (h *Held) Err() error { |
| 223 | h.mu.Lock() |
| 224 | defer h.mu.Unlock() |
| 225 | if h.spent { |
| 226 | return ErrSpent |
| 227 | } |
| 228 | if h.poison != nil { |
| 229 | return h.poison |
| 230 | } |
| 231 | return h.rn.Err() |
| 232 | } |
| 233 | |
| 234 | // Lost closes once on renewal failure or the safety deadline. Healthy transfer |
| 235 | // and explicit release do not close it. A worker must recheck Err/SafeUntil and |
| 236 | // prove its process tree stopped before the deadline; notification is not proof. |
| 237 | func (h *Held) Lost() <-chan struct{} { return h.rn.lost } |
| 238 | |
| 239 | // SafeUntil retains Go's monotonic clock component. It must not be serialized |
| 240 | // and later treated as ownership evidence; callers must keep checking health. |
| 241 | func (h *Held) SafeUntil() (time.Time, error) { |
| 242 | h.mu.Lock() |
| 243 | defer h.mu.Unlock() |
| 244 | if h.spent { |
| 245 | return time.Time{}, ErrSpent |
| 246 | } |
| 247 | if h.poison != nil { |
| 248 | return time.Time{}, h.poison |
| 249 | } |
| 250 | return h.rn.SafeUntil() |
| 251 | } |
| 252 | |
| 253 | func (h *Held) begin(release bool) error { |
| 254 | h.mu.Lock() |
| 255 | defer h.mu.Unlock() |
| 256 | if h.spent { |
| 257 | return ErrSpent |
| 258 | } |
| 259 | if h.busy { |
| 260 | return ErrBusy |
| 261 | } |
| 262 | if !release { |
| 263 | if h.poison != nil { |
| 264 | return h.poison |
| 265 | } |
| 266 | if err := h.rn.Err(); err != nil { |
| 267 | return err |
| 268 | } |
| 269 | } |
| 270 | h.busy = true |
| 271 | return nil |
| 272 | } |
| 273 | func (h *Held) finish(spent bool, poison error) { |
| 274 | h.mu.Lock() |
| 275 | defer h.mu.Unlock() |
| 276 | h.busy = false |
| 277 | h.spent = spent |
| 278 | if poison != nil { |
| 279 | h.poison = poison |
| 280 | } |
| 281 | } |
| 282 | |
| 283 | // Release is the only operation allowed after a failed checkpoint. A failed |
| 284 | // remote release still spends this handle; it cannot authorize more work. |
| 285 | func (h *Held) Release(ctx context.Context) error { |
| 286 | if err := h.begin(true); err != nil { |
| 287 | return err |
| 288 | } |
| 289 | defer h.finish(true, nil) |
| 290 | lease := h.rn.stop() |
| 291 | if lease == nil { |
| 292 | return nil |
| 293 | } |
| 294 | err := h.leaser.ReleaseLease(ctx, lease) |
| 295 | if err != nil && !releaseSettled(err) { |
| 296 | // The handle is spent, but the lease record may still stand: keep |
| 297 | // it so RetryRelease can delete it, on the same condition (#292). |
| 298 | h.mu.Lock() |
| 299 | h.owed = lease |
| 300 | h.mu.Unlock() |
| 301 | } |
| 302 | return err |
| 303 | } |
| 304 | |
| 305 | // releaseSettled reports whether a release error still leaves nothing of |
| 306 | // this lease to delete: the record is already gone, or its ETag no longer |
| 307 | // matches (the lease lapsed and is someone else's now). |
| 308 | func releaseSettled(err error) bool { |
| 309 | return errors.Is(err, lss3.ErrLeaseAlreadyReleased) || errors.Is(err, litestream.ErrLeaseNotHeld) |
| 310 | } |
| 311 | |
| 312 | // ReleaseOwed reports whether a failed Release left the lease record |
| 313 | // standing for RetryRelease. |
| 314 | func (h *Held) ReleaseOwed() bool { |
| 315 | h.mu.Lock() |
| 316 | defer h.mu.Unlock() |
| 317 | return h.owed != nil |
| 318 | } |
| 319 | |
| 320 | // ErrReleaseLapsed is RetryRelease's answer when the delete failed again |
| 321 | // but the owed record has passed its own expiry: the lease is free to any |
| 322 | // acquirer by then, so the obligation is retired rather than retried |
| 323 | // forever (#292). It wraps the delete's own error. |
| 324 | var ErrReleaseLapsed = errors.New("lease release retired: the lease record has passed its expiry") |
| 325 | |
| 326 | // RetryRelease discharges what a failed Release left owed (#292): the same |
| 327 | // conditional delete (If-Match on the lease's ETag), so it can never delete |
| 328 | // a lease someone else has taken since. nil once nothing is owed, including |
| 329 | // when the record is already gone or is no longer ours. A delete that fails |
| 330 | // again once the record has passed its own expiry retires the obligation |
| 331 | // with ErrReleaseLapsed: under a persistent refusal (AccessDenied, say) |
| 332 | // obligations would otherwise pile up and be retried forever. Callers do |
| 333 | // not run it concurrently with itself. |
| 334 | func (h *Held) RetryRelease(ctx context.Context) error { |
| 335 | h.mu.Lock() |
| 336 | lease := h.owed |
| 337 | h.mu.Unlock() |
| 338 | if lease == nil { |
| 339 | return nil |
| 340 | } |
| 341 | if err := h.leaser.ReleaseLease(ctx, lease); err != nil && !releaseSettled(err) { |
| 342 | if lease.ExpiresAt.IsZero() || time.Now().Before(lease.ExpiresAt) { |
| 343 | return err |
| 344 | } |
| 345 | h.mu.Lock() |
| 346 | if h.owed == lease { |
| 347 | h.owed = nil |
| 348 | } |
| 349 | h.mu.Unlock() |
| 350 | return fmt.Errorf("%w (expired %s): %w", ErrReleaseLapsed, lease.ExpiresAt.UTC().Format(time.RFC3339), err) |
| 351 | } |
| 352 | h.mu.Lock() |
| 353 | if h.owed == lease { |
| 354 | h.owed = nil |
| 355 | } |
| 356 | h.mu.Unlock() |
| 357 | return nil |
| 358 | } |
| 359 | |
| 360 | // Replicate trades the held lease for running replication of dbPath to the |
| 361 | // lease's S3 prefix. On success the Held is spent — renewal duty moves to |
| 362 | // the returned Replicator (Close it to release). On failure the lease is |
| 363 | // STILL HELD; the caller decides whether to retry or Release. |
| 364 | // |
| 365 | // syncInterval sets litestream's replica sync tick — the per-plan durability |
| 366 | // ladder's one knob. It is a MINIMUM spacing between S3 syncs (the monitor's |
| 367 | // change-notify only short-circuits the idle wait, never this spacing), so 5m |
| 368 | // bounds an awake host's hard-crash loss to ~5m and 24h to ~a day, while the |
| 369 | // wake-time initial pass and the sleep-time SyncAndWait/Close flush in full |
| 370 | // regardless — the interval never weakens hibernation's parked copy. 0 keeps |
| 371 | // litestream's 1s default (the stream class): idle ticks make zero S3 calls, |
| 372 | // so stream's request bill is change-driven, not clock-driven. |
| 373 | func (h *Held) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (*Replicator, error) { |
| 374 | if err := h.begin(false); err != nil { |
| 375 | return nil, err |
| 376 | } |
| 377 | transferred := false |
| 378 | defer func() { h.finish(transferred, nil) }() |
| 379 | opCtx, cancel := context.WithCancel(ctx) |
| 380 | stop := context.AfterFunc(h.rn.ctx, cancel) |
| 381 | defer stop() |
| 382 | defer cancel() |
| 383 | store, db, err := h.newStore(dbPath, syncInterval) |
| 384 | if err != nil { |
| 385 | return nil, err |
| 386 | } |
| 387 | if err := store.Open(opCtx); err != nil { |
| 388 | cancel() |
| 389 | closeErr := store.Close(opCtx) |
| 390 | return nil, errors.Join(fmt.Errorf("open litestream store: %w", err), closeErr) |
| 391 | } |
| 392 | if err := h.rn.Err(); err != nil { |
| 393 | cancel() |
| 394 | closeErr := store.Close(opCtx) |
| 395 | return nil, errors.Join(err, closeErr) |
| 396 | } |
| 397 | transferred = true |
| 398 | return &Replicator{leaser: h.leaser, store: store, db: db, rn: h.rn}, nil |
| 399 | } |
| 400 | |
| 401 | func (h *Held) newStore(dbPath string, syncInterval time.Duration) (*litestream.Store, *litestream.DB, error) { |
| 402 | db := litestream.NewDB(dbPath) |
| 403 | // Preserve the activator namespace: sharing the default metadata directory |
| 404 | // with another replicator corrupts shadow state even across sequential runs. |
| 405 | db.SetMetaPath(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream")) |
| 406 | db.MonitorInterval = dbMonitorInterval |
| 407 | client := h.replicaClient |
| 408 | if client == nil { |
| 409 | var err error |
| 410 | client, err = newReplicaClient(h.leaser.Bucket, h.leaser.Path, h.region) |
| 411 | if err != nil { |
| 412 | return nil, nil, err |
| 413 | } |
| 414 | } |
| 415 | db.Replica = litestream.NewReplicaWithClient(db, client) |
| 416 | if syncInterval > 0 { |
| 417 | db.Replica.SyncInterval = syncInterval |
| 418 | } |
| 419 | store := litestream.NewStore([]*litestream.DB{db}, compactionLevels) |
| 420 | store.SnapshotInterval = SnapshotInterval |
| 421 | store.SnapshotRetention = SnapshotRetention |
| 422 | if h.generationPrefix != "" { |
| 423 | // NewStore resets the DB flag. Set all three store/DB/compactor flags |
| 424 | // afterward, before Open starts any monitor or compaction work. |
| 425 | store.SetRetentionEnabled(false) |
| 426 | } |
| 427 | store.L0RetentionCheckInterval = l0RetentionCheckInterval |
| 428 | return store, db, nil |
| 429 | } |
| 430 | |
| 431 | // Replicator couples an exclusive S3 lease with a running litestream |
| 432 | // replication of one SQLite database to the same S3 prefix. The lease is |
| 433 | // acquired *before* replication starts and renewed on a ticker; Close stops |
| 434 | // replication and releases the lease. |
| 435 | type Replicator struct { |
| 436 | leaser *lss3.Leaser |
| 437 | store *litestream.Store |
| 438 | db *litestream.DB |
| 439 | rn *renewer |
| 440 | mu sync.Mutex |
| 441 | busy bool |
| 442 | spent bool |
| 443 | poison error |
| 444 | replicaClient litestream.ReplicaClient // private local-replica test seam |
| 445 | } |
| 446 | |
| 447 | // ReplicateWithLease acquires the single-writer lease at |
| 448 | // s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream |
| 449 | // replication of dbPath to s3://bucket/prefix at the default (stream-class) |
| 450 | // sync interval. If another process holds the lease it fails LOUDLY with a |
| 451 | // *HeldError (check with errors.As) — this is the signal |
| 452 | // the activator reacts to. |
| 453 | func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*Replicator, error) { |
| 454 | held, err := Acquire(ctx, bucket, prefix, region) |
| 455 | if err != nil { |
| 456 | return nil, err |
| 457 | } |
| 458 | r, err := held.Replicate(ctx, dbPath, 0) |
| 459 | if err != nil { |
| 460 | _ = held.Release(ctx) |
| 461 | return nil, err |
| 462 | } |
| 463 | return r, nil |
| 464 | } |
| 465 | |
| 466 | // SyncAndWait forces a full sync of the database to the replica and blocks |
| 467 | // until the remote position catches up — used by tests to make replication |
| 468 | // observable without waiting on monitor intervals. |
| 469 | func (r *Replicator) SyncAndWait(ctx context.Context) error { |
| 470 | return r.db.SyncAndWait(ctx) |
| 471 | } |
| 472 | |
| 473 | // Err reports whether lease renewal has failed (leadership lost). |
| 474 | func (r *Replicator) Err() error { return r.rn.Err() } |
| 475 | |
| 476 | func (r *Replicator) SafeUntil() (time.Time, error) { return r.rn.SafeUntil() } |
| 477 | |
| 478 | // Lost is the same loss notification channel held before replication began. |
| 479 | // Normal Close does not signal loss; SafeUntil refuses the stopped handle. |
| 480 | func (r *Replicator) Lost() <-chan struct{} { return r.rn.lost } |
| 481 | |
| 482 | // Abandon stops lease renewal and replication WITHOUT the final sync or the |
| 483 | // lease release — the wreckage a crashed box leaves behind: lock.json simply |
| 484 | // expires on its TTL, and the replica holds whatever was last synced. It |
| 485 | // exists so tests can simulate hard box death in-process; production code has |
| 486 | // no reason to call it (a real crash needs no help). |
| 487 | func (r *Replicator) Abandon(ctx context.Context) { |
| 488 | r.rn.stop() // deliberately no ReleaseLease — the TTL is the point |
| 489 | // litestream's only teardown is Close, which includes a sync a real crash |
| 490 | // would skip; it ships nothing as long as the caller killed the writer |
| 491 | // first, and reclaims the replication goroutines either way. |
| 492 | _ = r.store.Close(ctx) |
| 493 | } |
| 494 | |
| 495 | // Close stops replication (final sync included via litestream's DB.Close) |
| 496 | // and releases the lease so the next claimant can acquire immediately. |
| 497 | func (r *Replicator) Close(ctx context.Context) error { |
| 498 | // Force one full sync before teardown. litestream's DB.Close only |
| 499 | // final-syncs an *initialized* DB, and initialization happens on the |
| 500 | // monitor's first pass — so a checkpoint within ~a second of Replicate |
| 501 | // could silently ship nothing (a drain right after a wake, caught by the |
| 502 | // mesh live test: the replica held no restorable snapshot). SyncAndWait |
| 503 | // forces init + a replica sync; its error is deliberately dropped — with |
| 504 | // the DB now initialized, store.Close's own final sync is the one that |
| 505 | // counts, and its error is returned. |
| 506 | _ = r.db.SyncAndWait(ctx) |
| 507 | err := r.store.Close(ctx) |
| 508 | |
| 509 | lease := r.rn.stop() |
| 510 | if lease != nil { |
| 511 | // The release gets its OWN budget, never the caller's: a wedged final |
| 512 | // sync exhausts ctx, and releasing under a dead context leaves the |
| 513 | // lease held until TTL — the next wake then 503s for up to the full |
| 514 | // TTL over nothing (hit live on the first real instance flip). The |
| 515 | // release is |
| 516 | // one small S3 DELETE; five seconds is generous. |
| 517 | relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 518 | defer cancel() |
| 519 | if e := r.leaser.ReleaseLease(relCtx, lease); e != nil && err == nil { |
| 520 | err = e |
| 521 | } |
| 522 | } |
| 523 | return err |
| 524 | } |
| 525 | |
| 526 | // Restore pulls the newest replicated state of s3://bucket/prefix into |
| 527 | // dbPath via litestream's restore API — the cold-wake path when the local DB |
| 528 | // is missing. Returns restored=false with a nil error when the prefix holds |
| 529 | // no backup at all (a brand-new instance that has never replicated boots |
| 530 | // fresh instead). The caller must hold the prefix's lease: restoring outside |
| 531 | // it could resurrect a stale copy that later shadows the true holder's newer |
| 532 | // state. |
| 533 | func Restore(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error) { |
| 534 | // Resolved the same way the leaser's client resolves it — a restore that |
| 535 | // signed for a different region than the lease guarding it would be a |
| 536 | // silent split against a store that checks. |
| 537 | region, err := resolveRegion(ctx, region) |
| 538 | if err != nil { |
| 539 | return false, err |
| 540 | } |
| 541 | client, err := newReplicaClient(bucket, prefix, region) |
| 542 | if err != nil { |
| 543 | return false, err |
| 544 | } |
| 545 | |
| 546 | // A nested row's instance directory may not exist yet the first time it |
| 547 | // wakes: the directory is created by the spawn path (isolated-exec |
| 548 | // design §3.1), which runs AFTER this restore. Without this, a fresh |
| 549 | // nested row's very first wake fails with ENOENT before it ever gets to |
| 550 | // create anything. |
| 551 | if err := os.MkdirAll(filepath.Dir(dbPath), 0o700); err != nil { |
| 552 | return false, fmt.Errorf("create %s: %w", filepath.Dir(dbPath), err) |
| 553 | } |
| 554 | |
| 555 | // A nil *DB is litestream's own pattern for URL restores (the CLI does the |
| 556 | // same): Restore only needs the replica client and the output path. |
| 557 | r := litestream.NewReplicaWithClient(nil, client) |
| 558 | opt := litestream.NewRestoreOptions() |
| 559 | opt.OutputPath = dbPath |
| 560 | if err := r.Restore(ctx, opt); err != nil { |
| 561 | if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) { |
| 562 | return false, nil |
| 563 | } |
| 564 | return false, err |
| 565 | } |
| 566 | return true, nil |
| 567 | } |
| 568 | |
amadan