amadan
feat: lease — single-writer S3 lease + litestream replication (from Eleven)
infra: flagship skeleton — ie region, buckets, SSM-only host (apply queued for operator; blocked by permission classifier).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
8 files changed,
+874
−6
go.mod+10 −6go.sum+4 −0infra/flagship/host-init.sh+31 −0infra/flagship/main.tf+85 −0infra/flagship/outputs.tf+15 −0infra/flagship/region-ie.tf+115 −0internal/lease/lease.go+379 −0internal/lease/lease_test.go+235 −0
diff --git a/go.mod b/go.mod| index 5786921e2..22ac241e8 100644 |
| --- a/go.mod |
| +++ b/go.mod |
| @@ -2,14 +2,20 @@ module github.com/carlosframework/platform |
| go 1.26 |
| -require modernc.org/sqlite v1.55.0 |
| +require ( |
| + github.com/aws/aws-sdk-go-v2 v1.42.1 |
| + github.com/aws/aws-sdk-go-v2/config v1.32.6 |
| + github.com/aws/aws-sdk-go-v2/service/s3 v1.105.0 |
| + github.com/benbjohnson/litestream v0.5.14 |
| + golang.org/x/crypto v0.52.0 |
| + modernc.org/sqlite v1.55.0 |
| +) |
| require ( |
| - github.com/aws/aws-sdk-go-v2 v1.42.1 // indirect |
| github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14 // indirect |
| - github.com/aws/aws-sdk-go-v2/config v1.32.6 // indirect |
| github.com/aws/aws-sdk-go-v2/credentials v1.19.6 // indirect |
| github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 // indirect |
| + github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18 // indirect |
| github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30 // indirect |
| github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.30 // indirect |
| github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect |
| @@ -18,13 +24,11 @@ require ( |
| github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23 // indirect |
| github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.30 // indirect |
| github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31 // indirect |
| - github.com/aws/aws-sdk-go-v2/service/s3 v1.105.0 // indirect |
| github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 // indirect |
| github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 // indirect |
| github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect |
| github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect |
| github.com/aws/smithy-go v1.27.3 // indirect |
| - github.com/benbjohnson/litestream v0.5.14 // indirect |
| github.com/beorn7/perks v1.0.1 // indirect |
| github.com/cespare/xxhash/v2 v2.3.0 // indirect |
| github.com/dustin/go-humanize v1.0.1 // indirect |
| @@ -49,7 +53,7 @@ require ( |
| github.com/superfly/ltx v0.5.1 // indirect |
| github.com/tetratelabs/wazero v1.2.1 // indirect |
| github.com/wasilibs/go-re2 v1.3.0 // indirect |
| - golang.org/x/crypto v0.52.0 // indirect |
| + golang.org/x/net v0.55.0 // indirect |
| golang.org/x/sync v0.21.0 // indirect |
| golang.org/x/sys v0.46.0 // indirect |
| golang.org/x/text v0.37.0 // indirect |
diff --git a/go.sum b/go.sum| index 686347c92..0d5615ed3 100644 |
| --- a/go.sum |
| +++ b/go.sum |
| @@ -8,6 +8,8 @@ github.com/aws/aws-sdk-go-v2/credentials v1.19.6 h1:F9vWao2TwjV2MyiyVS+duza0NIRt |
| github.com/aws/aws-sdk-go-v2/credentials v1.19.6/go.mod h1:SgHzKjEVsdQr6Opor0ihgWtkWdfRAIwxYzSJ8O85VHY= |
| github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 h1:80+uETIWS1BqjnN9uJ0dBUaETh+P1XwFy5vwHwK5r9k= |
| github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16/go.mod h1:wOOsYuxYuB/7FlnVtzeBYRcjSRtQpAW0hCP7tIULMwo= |
| +github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18 h1:9vWXHtaepwoAl/UuKzxwgOoJDXPCC3hvgNMfcmdS2Tk= |
| +github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18/go.mod h1:sKuUZ+MwUTuJbYvZ8pK0x10LvgcJK3Y4rmh63YBekwk= |
| github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30 h1:xM/Is9cKMHa8Jj8zkvWhvrFkZsXJV9E+BB4g0HW0duQ= |
| github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30/go.mod h1:WueJeNDZvK1fMYEWJIkcivBfEzUkTpBhzlrUKKY8EuA= |
| github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.30 h1:jn46zC9LdsVR/ZpMIJqMqb8hHv31BlLx3ulVqNspUOk= |
| @@ -94,6 +96,8 @@ golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= |
| golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= |
| golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= |
| golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= |
| +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= |
| +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= |
| golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= |
| golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= |
| golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
diff --git a/infra/flagship/host-init.sh b/infra/flagship/host-init.sh| new file mode 100644 |
| index 000000000..5d97c61f9 |
| --- /dev/null |
| +++ b/infra/flagship/host-init.sh |
| @@ -0,0 +1,31 @@ |
| +#!/bin/bash |
| +# 🤖 First-boot init for a carlos flagship host (Debian arm64). Deliberately |
| +# minimal: the carlos binary arrives via bootstrap after the box is up — |
| +# this script only prepares the ground the binary expects. |
| +set -euo pipefail |
| + |
| +# The service user and the three directories every carlos host has. |
| +useradd --system --home /opt/carlos --shell /usr/sbin/nologin carlos || true |
| +mkdir -p /data /opt/carlos/versions /opt/carlos/live /etc/carlos |
| +chown -R carlos:carlos /data /opt/carlos |
| +chmod 0700 /data |
| + |
| +# /run/carlos is tmpfs-backed; recreate on every boot. |
| +cat >/etc/tmpfiles.d/carlos.conf <<'EOF' |
| +d /run/carlos 0755 carlos carlos - |
| +EOF |
| +systemd-tmpfiles --create /etc/tmpfiles.d/carlos.conf |
| + |
| +# Keep the OS patched; everything else ships through carlos channels. |
| +export DEBIAN_FRONTEND=noninteractive |
| +apt-get update -q |
| +apt-get install -qy unattended-upgrades curl awscli |
| +systemctl enable --now unattended-upgrades |
| + |
| +# Admin access is SSM Session Manager only — no SSH ingress exists. |
| +curl -fsSL -o /tmp/amazon-ssm-agent.deb \ |
| + "https://s3.eu-west-1.amazonaws.com/amazon-ssm-eu-west-1/latest/debian_arm64/amazon-ssm-agent.deb" |
| +dpkg -i /tmp/amazon-ssm-agent.deb |
| +systemctl enable --now amazon-ssm-agent |
| + |
| +echo "carlos host ready — awaiting bootstrap" >/etc/motd |
diff --git a/infra/flagship/main.tf b/infra/flagship/main.tf| new file mode 100644 |
| index 000000000..46f74c278 |
| --- /dev/null |
| +++ b/infra/flagship/main.tf |
| @@ -0,0 +1,85 @@ |
| +# Flagship CARLOS deployment — AWS account 271376211898 (adopted from |
| +# Keymail; platform spec §12). Tofu owns boxes, buckets, IAM, and static |
| +# DNS; the platform owns apps, releases, instances, and dynamic DNS. |
| +# Nothing account-shaped (tenant hostnames, routes) may ever enter this |
| +# state — `tofu apply` must always be safe. |
| + |
| +terraform { |
| + required_version = ">= 1.10" |
| + backend "s3" { |
| + bucket = "carlos-tfstate-271376211898" |
| + key = "flagship/terraform.tfstate" |
| + region = "eu-west-1" |
| + profile = "keymail" |
| + use_lockfile = true |
| + } |
| + required_providers { |
| + aws = { |
| + source = "hashicorp/aws" |
| + version = "~> 6.0" |
| + } |
| + } |
| +} |
| + |
| +provider "aws" { |
| + region = "eu-west-1" |
| + profile = "keymail" |
| +} |
| + |
| +locals { |
| + account = "271376211898" |
| +} |
| + |
| +# The deployment bucket: apps, releases, channels, instance records, host |
| +# config. Routing metadata only — nothing customer-shaped rests here. |
| +resource "aws_s3_bucket" "deployment" { |
| + bucket = "carlos-flagship-${local.account}" |
| + lifecycle { prevent_destroy = true } |
| +} |
| + |
| +# The IE regional data bucket: parked databases, WAL streams, leases, |
| +# blobs. This bucket is the residency promise. |
| +resource "aws_s3_bucket" "data_ie" { |
| + bucket = "carlos-data-ie-${local.account}" |
| + lifecycle { prevent_destroy = true } |
| +} |
| + |
| +resource "aws_s3_bucket_versioning" "deployment" { |
| + bucket = aws_s3_bucket.deployment.id |
| + versioning_configuration { status = "Enabled" } |
| +} |
| + |
| +resource "aws_s3_bucket_versioning" "data_ie" { |
| + bucket = aws_s3_bucket.data_ie.id |
| + versioning_configuration { status = "Enabled" } |
| +} |
| + |
| +resource "aws_s3_bucket_public_access_block" "deployment" { |
| + bucket = aws_s3_bucket.deployment.id |
| + block_public_acls = true |
| + ignore_public_acls = true |
| + block_public_policy = true |
| + restrict_public_buckets = true |
| +} |
| + |
| +resource "aws_s3_bucket_public_access_block" "data_ie" { |
| + bucket = aws_s3_bucket.data_ie.id |
| + block_public_acls = true |
| + ignore_public_acls = true |
| + block_public_policy = true |
| + restrict_public_buckets = true |
| +} |
| + |
| +resource "aws_s3_bucket_server_side_encryption_configuration" "deployment" { |
| + bucket = aws_s3_bucket.deployment.id |
| + rule { |
| + apply_server_side_encryption_by_default { sse_algorithm = "AES256" } |
| + } |
| +} |
| + |
| +resource "aws_s3_bucket_server_side_encryption_configuration" "data_ie" { |
| + bucket = aws_s3_bucket.data_ie.id |
| + rule { |
| + apply_server_side_encryption_by_default { sse_algorithm = "AES256" } |
| + } |
| +} |
diff --git a/infra/flagship/outputs.tf b/infra/flagship/outputs.tf| new file mode 100644 |
| index 000000000..8aca3bdb6 |
| --- /dev/null |
| +++ b/infra/flagship/outputs.tf |
| @@ -0,0 +1,15 @@ |
| +output "ie_eip" { |
| + value = aws_eip.host_ie.public_ip |
| +} |
| + |
| +output "deployment_bucket" { |
| + value = aws_s3_bucket.deployment.bucket |
| +} |
| + |
| +output "data_bucket_ie" { |
| + value = aws_s3_bucket.data_ie.bucket |
| +} |
| + |
| +output "host_role_arn" { |
| + value = aws_iam_role.host.arn |
| +} |
diff --git a/infra/flagship/region-ie.tf b/infra/flagship/region-ie.tf| new file mode 100644 |
| index 000000000..49a741d91 |
| --- /dev/null |
| +++ b/infra/flagship/region-ie.tf |
| @@ -0,0 +1,115 @@ |
| +# The IE region of the flagship: one Graviton box running `carlos agent`, |
| +# behind an Elastic IP. The box is disposable by construction — all |
| +# durable state is in the buckets — so there is no backup story for hosts |
| +# at all, only for buckets. |
| + |
| +data "aws_ami" "debian_arm64" { |
| + most_recent = true |
| + owners = ["136693071363"] # Debian |
| + filter { |
| + name = "name" |
| + values = ["debian-13-arm64-*"] |
| + } |
| + filter { |
| + name = "architecture" |
| + values = ["arm64"] |
| + } |
| +} |
| + |
| +resource "aws_security_group" "host_ie" { |
| + name = "carlos-flagship-ie" |
| + description = "carlos flagship IE host" |
| + ingress { |
| + from_port = 80 |
| + to_port = 80 |
| + protocol = "tcp" |
| + cidr_blocks = ["0.0.0.0/0"] |
| + } |
| + ingress { |
| + from_port = 443 |
| + to_port = 443 |
| + protocol = "tcp" |
| + cidr_blocks = ["0.0.0.0/0"] |
| + } |
| + # No SSH ingress: admin access is SSM Session Manager only, and the |
| + # carlos binary arrives from the deployment bucket, not over scp. |
| + egress { |
| + from_port = 0 |
| + to_port = 0 |
| + protocol = "-1" |
| + cidr_blocks = ["0.0.0.0/0"] |
| + } |
| +} |
| + |
| +resource "aws_iam_role" "host" { |
| + name = "carlos-flagship-host" |
| + assume_role_policy = jsonencode({ |
| + Version = "2012-10-17" |
| + Statement = [{ |
| + Effect = "Allow" |
| + Principal = { Service = "ec2.amazonaws.com" } |
| + Action = "sts:AssumeRole" |
| + }] |
| + }) |
| +} |
| + |
| +# Residency enforced by construction: the host role reaches the deployment |
| +# bucket plus ITS OWN region's data bucket and nothing else. |
| +resource "aws_iam_role_policy" "host_buckets" { |
| + name = "carlos-flagship-host-buckets" |
| + role = aws_iam_role.host.id |
| + policy = jsonencode({ |
| + Version = "2012-10-17" |
| + Statement = [ |
| + { |
| + Effect = "Allow" |
| + Action = ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"] |
| + Resource = [ |
| + "${aws_s3_bucket.deployment.arn}/*", |
| + "${aws_s3_bucket.data_ie.arn}/*", |
| + ] |
| + }, |
| + { |
| + Effect = "Allow" |
| + Action = ["s3:ListBucket", "s3:GetBucketLocation"] |
| + Resource = [aws_s3_bucket.deployment.arn, aws_s3_bucket.data_ie.arn] |
| + }, |
| + { |
| + Effect = "Allow" |
| + Action = ["ssm:GetParameter"] |
| + Resource = "arn:aws:ssm:eu-west-1:${local.account}:parameter/carlos/*" |
| + }, |
| + ] |
| + }) |
| +} |
| + |
| +resource "aws_iam_role_policy_attachment" "host_ssm" { |
| + role = aws_iam_role.host.name |
| + policy_arn = "arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore" |
| +} |
| + |
| +resource "aws_iam_instance_profile" "host" { |
| + name = "carlos-flagship-host" |
| + role = aws_iam_role.host.name |
| +} |
| + |
| +resource "aws_instance" "host_ie" { |
| + ami = data.aws_ami.debian_arm64.id |
| + instance_type = "t4g.small" |
| + vpc_security_group_ids = [aws_security_group.host_ie.id] |
| + iam_instance_profile = aws_iam_instance_profile.host.name |
| + user_data = file("${path.module}/host-init.sh") |
| + user_data_replace_on_change = true |
| + root_block_device { |
| + volume_size = 20 |
| + volume_type = "gp3" |
| + encrypted = true |
| + } |
| + tags = { Name = "carlos-flagship-ie" } |
| +} |
| + |
| +resource "aws_eip" "host_ie" { |
| + domain = "vpc" |
| + instance = aws_instance.host_ie.id |
| + tags = { Name = "carlos-flagship-ie" } |
| +} |
diff --git a/internal/lease/lease.go b/internal/lease/lease.go| new file mode 100644 |
| index 000000000..57c6d99a4 |
| --- /dev/null |
| +++ b/internal/lease/lease.go |
| @@ -0,0 +1,379 @@ |
| +// Ported from lchat internal/activator/replica.go. |
| +// |
| +// Package lease holds the replication primitives for instance |
| +// hibernation (issue #550 Part 4 v2): litestream embedded as a library — pure |
| +// modernc.org/sqlite, no cgo — with its S3 lease (lock.json via conditional |
| +// writes) wrapped around replication as a single-writer lock. The lease is |
| +// the truth of ownership: a second claimant fails loudly with a typed |
| +// *HeldError (the CLI, by contrast, silently absorbs a |
| +// second replicator — proven on #550), and a failed renewal means leadership |
| +// is lost and the caller must stand down. The activator (wake/sleep) |
| +// builds on these primitives. |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "path/filepath" |
| + "sync" |
| + "time" |
| + |
| + awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| + "github.com/benbjohnson/litestream" |
| + lss3 "github.com/benbjohnson/litestream/s3" |
| +) |
| + |
| +// LeaseTTL is deliberately long — 30× litestream's 30s default. Renewal runs |
| +// at TTL/3 and each renewal is one S3 Tier-1 PUT, so at the old 30s TTL the |
| +// lease heartbeat alone was 8,640 PUTs/day per awake instance (~$16/yr — |
| +// ~70% of the fleet's entire S3 bill, and more than a €5/y space's revenue); |
| +// at 15m it's 288/day (~$0.50/yr). The price is the failover window: a |
| +// HARD-crashed holder (kill -9, box death) strands the lease until expiry, |
| +// so a peer's takeover — or a re-wake on the same box — can stall up to |
| +// ~15m. Every graceful path (sleep, deploy, shutdown) releases the lease |
| +// explicitly and is unaffected. Shorten this only if fast crash-failover |
| +// ever outranks per-instance cost. |
| +const LeaseTTL = 15 * time.Minute |
| + |
| +// compactionLevels replaces litestream.DefaultCompactionLevels (L1 30s / |
| +// L2 5m / L3 1h) with a far gentler ladder. The defaults are tuned for a |
| +// long-lived replicator where background compaction promptness matters; a |
| +// hibernating instance instead checkpoints deterministically at sleep via |
| +// SyncAndWait (which calls db.Sync + Replica.Sync directly, independent of |
| +// every interval here), so aggressive background compaction buys nothing — |
| +// it only burns S3 Tier-1 LISTs: each level's monitor tick costs up to two |
| +// LISTs per DB even with zero writes — L1 at 30s dominated the fleet's idle |
| +// request bill, and the first fix (5m/30m/3h) still left ~1.3k LISTs/day per |
| +// awake instance once the lease PUTs were tamed. |
| +// Correctness is untouched — compaction is purely a restore-time |
| +// read-amplification optimisation (restore replays snapshot + whatever L0 |
| +// files exist), and each level's monitor still runs one immediate pass at |
| +// store.Open, so every wake compacts anything a previous short awake window |
| +// left pending. Same three-level geometric shape, so restore reads stay |
| +// bounded: worst case a hard crash replays up to ~an hour of L0 files. |
| +var compactionLevels = litestream.CompactionLevels{ |
| + {Level: 0, Interval: 0}, |
| + {Level: 1, Interval: 1 * time.Hour}, |
| + {Level: 2, Interval: 6 * time.Hour}, |
| + {Level: 3, Interval: 24 * time.Hour}, |
| +} |
| + |
| +// dbMonitorInterval slows litestream's DB monitor (default 1s) — the local |
| +// WAL poll that shovels new frames into shadow LTX. It makes no S3 calls, |
| +// but 1s polling per awake instance is pointless churn when the sleep-time |
| +// SyncAndWait ships everything regardless; 10s bounds an awake instance's |
| +// crash-loss window to ~10s, plenty tight for chat. |
| +// (Replica.SyncInterval stays at its 1s default on purpose: idle replica |
| +// ticks make zero S3 calls — the upload loop no-ops when the local TXID |
| +// hasn't moved — and while awake it bounds crash-loss to ~1s.) |
| +const dbMonitorInterval = 10 * time.Second |
| + |
| +// l0RetentionCheckInterval slows the store's L0 retention monitor (default |
| +// 15s — tuned for litestream's VFS read replicas, which we don't run). Every |
| +// tick LISTs L1 and usually L0 per DB even with zero writes — the single |
| +// biggest idle-request source. L0 files only become reclaimable after an L1 |
| +// compaction, so checking more often than the L1 interval (1h above) is pure |
| +// waste; matching it cuts the LISTs 240× vs the default with no change to |
| +// what gets deleted. |
| +const l0RetentionCheckInterval = 1 * time.Hour |
| + |
| +// HeldError reports that another process holds the single-writer lease. It |
| +// wraps litestream's *LeaseExistsError so callers of this package never need |
| +// to import litestream to react to a contested lease (check with errors.As). |
| +type HeldError struct { |
| + Owner string |
| + ExpiresAt time.Time |
| +} |
| + |
| +func (e *HeldError) Error() string { |
| + if e.Owner != "" { |
| + return fmt.Sprintf("lease already held by %s until %s", e.Owner, e.ExpiresAt.Format(time.RFC3339)) |
| + } |
| + return fmt.Sprintf("lease already held until %s", e.ExpiresAt.Format(time.RFC3339)) |
| +} |
| + |
| +// renewer keeps a held lease alive on a ticker. A failed renewal is recorded |
| +// and renewal stops — leadership is lost and the owner must stand down; |
| +// retrying could fence with a new claimant who took over after expiry. |
| +type renewer struct { |
| + leaser *lss3.Leaser |
| + |
| + mu sync.Mutex |
| + lease *litestream.Lease |
| + err error // set if a renewal ever fails (leadership lost) |
| + |
| + cancel context.CancelFunc |
| + done chan struct{} |
| +} |
| + |
| +func startRenewer(leaser *lss3.Leaser, lease *litestream.Lease) *renewer { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + rn := &renewer{leaser: leaser, lease: lease, cancel: cancel, done: make(chan struct{})} |
| + go func() { |
| + defer close(rn.done) |
| + t := time.NewTicker(LeaseTTL / 3) |
| + defer t.Stop() |
| + for { |
| + select { |
| + case <-ctx.Done(): |
| + return |
| + case <-t.C: |
| + rn.mu.Lock() |
| + cur := rn.lease |
| + rn.mu.Unlock() |
| + renewed, err := rn.leaser.RenewLease(ctx, cur) |
| + rn.mu.Lock() |
| + if err != nil { |
| + rn.err = err |
| + } else { |
| + rn.lease = renewed |
| + } |
| + rn.mu.Unlock() |
| + if err != nil { |
| + return |
| + } |
| + } |
| + } |
| + }() |
| + return rn |
| +} |
| + |
| +// Err reports whether a renewal has failed (leadership lost). |
| +func (rn *renewer) Err() error { |
| + rn.mu.Lock() |
| + defer rn.mu.Unlock() |
| + return rn.err |
| +} |
| + |
| +// stop halts renewal and returns the most recent lease (for release, or to |
| +// hand renewal duty to a successor renewer). |
| +func (rn *renewer) stop() *litestream.Lease { |
| + rn.cancel() |
| + <-rn.done |
| + rn.mu.Lock() |
| + defer rn.mu.Unlock() |
| + return rn.lease |
| +} |
| + |
| +// Held is an acquired single-writer lease that is not yet coupled to |
| +// replication. The wake path needs this split: the DB restore must happen |
| +// *under* the lock (restoring without it could later shadow newer S3 state |
| +// written by the true holder), but replication can only start once the DB |
| +// file exists. Renewal runs from acquisition, so a slow restore can't silently |
| +// outlive the TTL. |
| +type Held struct { |
| + leaser *lss3.Leaser |
| + region string // the leaser's client keeps its region private; carry it for Replicate |
| + rn *renewer |
| +} |
| + |
| +// Acquire claims the single-writer lease at s3://bucket/prefix/lock.json |
| +// and starts renewing it. If another process holds the lease it fails LOUDLY |
| +// with a *HeldError (check with errors.As) — this is the |
| +// signal the activator reacts to. Region may be empty; the AWS default chain |
| +// (env/profile) resolves it. |
| +func Acquire(ctx context.Context, bucket, prefix, region string) (*Held, error) { |
| + // The leaser wants a ready-made S3 client (unlike the replica client, |
| + // which builds its own). Default AWS credential chain. |
| + var opts []func(*awsconfig.LoadOptions) error |
| + if region != "" { |
| + opts = append(opts, awsconfig.WithRegion(region)) |
| + } |
| + cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...) |
| + if err != nil { |
| + return nil, fmt.Errorf("load aws config: %w", err) |
| + } |
| + |
| + leaser := lss3.NewLeaser() |
| + leaser.SetClient(awss3.NewFromConfig(cfg)) |
| + leaser.Bucket = bucket |
| + leaser.Path = prefix // lock lives at <prefix>/lock.json, colocated with the replica |
| + leaser.TTL = LeaseTTL |
| + |
| + lease, err := leaser.AcquireLease(ctx) |
| + if err != nil { |
| + // A *litestream.LeaseExistsError means someone else holds it — |
| + // re-surfaced as *HeldError so callers stay litestream-free. |
| + var lexists *litestream.LeaseExistsError |
| + if errors.As(err, &lexists) { |
| + return nil, fmt.Errorf("acquire lease: %w", &HeldError{Owner: lexists.Owner, ExpiresAt: lexists.ExpiresAt}) |
| + } |
| + return nil, fmt.Errorf("acquire lease: %w", err) |
| + } |
| + return &Held{leaser: leaser, region: region, rn: startRenewer(leaser, lease)}, nil |
| +} |
| + |
| +// Err reports whether lease renewal has failed (leadership lost). |
| +func (h *Held) Err() error { return h.rn.Err() } |
| + |
| +// Release stops renewal and releases the lease — the bail-out when a wake |
| +// fails between acquisition and replication. |
| +func (h *Held) Release(ctx context.Context) error { |
| + lease := h.rn.stop() |
| + if lease == nil { |
| + return nil |
| + } |
| + return h.leaser.ReleaseLease(ctx, lease) |
| +} |
| + |
| +// Replicate trades the held lease for running replication of dbPath to the |
| +// lease's S3 prefix. On success the Held is spent — renewal duty moves to |
| +// the returned Replicator (Close it to release). On failure the lease is |
| +// STILL HELD; the caller decides whether to retry or Release. |
| +// |
| +// syncInterval sets litestream's replica sync tick — the per-plan durability |
| +// ladder's one knob. It is a MINIMUM spacing between S3 syncs (the monitor's |
| +// change-notify only short-circuits the idle wait, never this spacing), so 5m |
| +// bounds an awake host's hard-crash loss to ~5m and 24h to ~a day, while the |
| +// wake-time initial pass and the sleep-time SyncAndWait/Close flush in full |
| +// regardless — the interval never weakens hibernation's parked copy. 0 keeps |
| +// litestream's 1s default (the stream class): idle ticks make zero S3 calls, |
| +// so stream's request bill is change-driven, not clock-driven. |
| +func (h *Held) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (*Replicator, error) { |
| + if err := h.rn.Err(); err != nil { |
| + return nil, fmt.Errorf("lease already lost: %w", err) |
| + } |
| + |
| + db := litestream.NewDB(dbPath) |
| + // The activator's replication state lives in its OWN meta directory — |
| + // litestream derives ".<db>-litestream" beside the DB by default, which any |
| + // other litestream process over the same file (the box-wide backup |
| + // replicator, before or after a handover) also uses. Sharing corrupts LTX |
| + // state even SEQUENTIALLY: the fleet-wide flip left a dozen hosts in an |
| + // error loop until their meta dirs were cleared by hand. A distinct |
| + // namespace ends the class. |
| + db.SetMetaPath(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream")) |
| + db.MonitorInterval = dbMonitorInterval |
| + client := lss3.NewReplicaClient() |
| + client.Bucket = h.leaser.Bucket |
| + client.Path = h.leaser.Path |
| + client.Region = h.region |
| + db.Replica = litestream.NewReplicaWithClient(db, client) |
| + if syncInterval > 0 { |
| + db.Replica.SyncInterval = syncInterval |
| + } |
| + |
| + store := litestream.NewStore([]*litestream.DB{db}, compactionLevels) |
| + store.L0RetentionCheckInterval = l0RetentionCheckInterval |
| + if err := store.Open(ctx); err != nil { |
| + return nil, fmt.Errorf("open litestream store: %w", err) |
| + } |
| + |
| + // Hand renewal to the replicator. The handoff gap between stopping one |
| + // renewer and starting the next is microseconds against a 30s TTL. |
| + lease := h.rn.stop() |
| + return &Replicator{leaser: h.leaser, store: store, db: db, rn: startRenewer(h.leaser, lease)}, nil |
| +} |
| + |
| +// Replicator couples an exclusive S3 lease with a running litestream |
| +// replication of one SQLite database to the same S3 prefix. The lease is |
| +// acquired *before* replication starts and renewed on a ticker; Close stops |
| +// replication and releases the lease. |
| +type Replicator struct { |
| + leaser *lss3.Leaser |
| + store *litestream.Store |
| + db *litestream.DB |
| + rn *renewer |
| +} |
| + |
| +// ReplicateWithLease acquires the single-writer lease at |
| +// s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream |
| +// replication of dbPath to s3://bucket/prefix at the default (stream-class) |
| +// sync interval. If another process holds the lease it fails LOUDLY with a |
| +// *HeldError (check with errors.As) — this is the signal |
| +// the activator reacts to. |
| +func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*Replicator, error) { |
| + held, err := Acquire(ctx, bucket, prefix, region) |
| + if err != nil { |
| + return nil, err |
| + } |
| + r, err := held.Replicate(ctx, dbPath, 0) |
| + if err != nil { |
| + _ = held.Release(ctx) |
| + return nil, err |
| + } |
| + return r, nil |
| +} |
| + |
| +// SyncAndWait forces a full sync of the database to the replica and blocks |
| +// until the remote position catches up — used by tests to make replication |
| +// observable without waiting on monitor intervals. |
| +func (r *Replicator) SyncAndWait(ctx context.Context) error { |
| + return r.db.SyncAndWait(ctx) |
| +} |
| + |
| +// Err reports whether lease renewal has failed (leadership lost). |
| +func (r *Replicator) Err() error { return r.rn.Err() } |
| + |
| +// Abandon stops lease renewal and replication WITHOUT the final sync or the |
| +// lease release — the wreckage a crashed box leaves behind: lock.json simply |
| +// expires on its TTL, and the replica holds whatever was last synced. It |
| +// exists so tests can simulate hard box death in-process; production code has |
| +// no reason to call it (a real crash needs no help). |
| +func (r *Replicator) Abandon(ctx context.Context) { |
| + r.rn.stop() // deliberately no ReleaseLease — the TTL is the point |
| + // litestream's only teardown is Close, which includes a sync a real crash |
| + // would skip; it ships nothing as long as the caller killed the writer |
| + // first, and reclaims the replication goroutines either way. |
| + _ = r.store.Close(ctx) |
| +} |
| + |
| +// Close stops replication (final sync included via litestream's DB.Close) |
| +// and releases the lease so the next claimant can acquire immediately. |
| +func (r *Replicator) Close(ctx context.Context) error { |
| + // Force one full sync before teardown. litestream's DB.Close only |
| + // final-syncs an *initialized* DB, and initialization happens on the |
| + // monitor's first pass — so a checkpoint within ~a second of Replicate |
| + // could silently ship nothing (a drain right after a wake, caught by the |
| + // mesh live test: the replica held no restorable snapshot). SyncAndWait |
| + // forces init + a replica sync; its error is deliberately dropped — with |
| + // the DB now initialized, store.Close's own final sync is the one that |
| + // counts, and its error is returned. |
| + _ = r.db.SyncAndWait(ctx) |
| + err := r.store.Close(ctx) |
| + |
| + lease := r.rn.stop() |
| + if lease != nil { |
| + // The release gets its OWN budget, never the caller's: a wedged final |
| + // sync exhausts ctx, and releasing under a dead context leaves the |
| + // lease held until TTL — the next wake then 503s for up to the full |
| + // TTL over nothing (hit live on the first real instance flip). The |
| + // release is |
| + // one small S3 DELETE; five seconds is generous. |
| + relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + if e := r.leaser.ReleaseLease(relCtx, lease); e != nil && err == nil { |
| + err = e |
| + } |
| + } |
| + return err |
| +} |
| + |
| +// Restore pulls the newest replicated state of s3://bucket/prefix into |
| +// dbPath via litestream's restore API — the cold-wake path when the local DB |
| +// is missing. Returns restored=false with a nil error when the prefix holds |
| +// no backup at all (a brand-new instance that has never replicated boots |
| +// fresh instead). The caller must hold the prefix's lease: restoring outside |
| +// it could resurrect a stale copy that later shadows the true holder's newer |
| +// state. |
| +func Restore(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error) { |
| + client := lss3.NewReplicaClient() |
| + client.Bucket = bucket |
| + client.Path = prefix |
| + client.Region = region |
| + |
| + // A nil *DB is litestream's own pattern for URL restores (the CLI does the |
| + // same): Restore only needs the replica client and the output path. |
| + r := litestream.NewReplicaWithClient(nil, client) |
| + opt := litestream.NewRestoreOptions() |
| + opt.OutputPath = dbPath |
| + if err := r.Restore(ctx, opt); err != nil { |
| + if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) { |
| + return false, nil |
| + } |
| + return false, err |
| + } |
| + return true, nil |
| +} |
diff --git a/internal/lease/lease_test.go b/internal/lease/lease_test.go| new file mode 100644 |
| index 000000000..299f5737a |
| --- /dev/null |
| +++ b/internal/lease/lease_test.go |
| @@ -0,0 +1,235 @@ |
| +// Ported from lchat internal/activator/replica_test.go. |
| +// |
| +// SPIKE tests (issue #550 Part 4 v2). Live S3 tests are gated on |
| +// CARLOS_TEST_S3_BUCKET; without it everything skips so `go test ./...` |
| +// stays green on CI and dev machines with no AWS credentials. |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "database/sql" |
| + "encoding/hex" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" |
| + |
| + _ "modernc.org/sqlite" |
| +) |
| + |
| +func liveConfig(t *testing.T) (bucket, prefix, region string) { |
| + t.Helper() |
| + bucket = os.Getenv("CARLOS_TEST_S3_BUCKET") |
| + if bucket == "" { |
| + t.Skip("CARLOS_TEST_S3_BUCKET not set; skipping live S3 spike test") |
| + } |
| + region = os.Getenv("CARLOS_TEST_S3_REGION") |
| + if region == "" { |
| + region = "eu-west-1" |
| + } |
| + b := make([]byte, 4) |
| + _, _ = rand.Read(b) |
| + prefix = fmt.Sprintf("spike/%d-%s", time.Now().Unix(), hex.EncodeToString(b)) |
| + t.Cleanup(func() { cleanupPrefix(t, bucket, prefix, region) }) |
| + return bucket, prefix, region |
| +} |
| + |
| +// seedDB creates a WAL-mode SQLite database via modernc.org/sqlite (the |
| +// driver this repo already uses — no cgo anywhere in the test). |
| +func seedDB(t *testing.T, path string) { |
| + t.Helper() |
| + db, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatalf("open sqlite: %v", err) |
| + } |
| + defer db.Close() |
| + if _, err := db.Exec(`PRAGMA journal_mode=WAL`); err != nil { |
| + t.Fatalf("wal mode: %v", err) |
| + } |
| + if _, err := db.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil { |
| + t.Fatalf("create table: %v", err) |
| + } |
| + for i := 0; i < 50; i++ { |
| + if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-%d", i)); err != nil { |
| + t.Fatalf("insert: %v", err) |
| + } |
| + } |
| +} |
| + |
| +func s3Client(t *testing.T, region string) *awss3.Client { |
| + t.Helper() |
| + cfg, err := awsconfig.LoadDefaultConfig(context.Background(), awsconfig.WithRegion(region)) |
| + if err != nil { |
| + t.Fatalf("aws config: %v", err) |
| + } |
| + return awss3.NewFromConfig(cfg) |
| +} |
| + |
| +func listKeys(t *testing.T, bucket, prefix, region string) []string { |
| + t.Helper() |
| + client := s3Client(t, region) |
| + var keys []string |
| + p := awss3.NewListObjectsV2Paginator(client, &awss3.ListObjectsV2Input{ |
| + Bucket: aws.String(bucket), |
| + Prefix: aws.String(prefix), |
| + }) |
| + for p.HasMorePages() { |
| + out, err := p.NextPage(context.Background()) |
| + if err != nil { |
| + t.Fatalf("list objects: %v", err) |
| + } |
| + for _, o := range out.Contents { |
| + keys = append(keys, *o.Key) |
| + } |
| + } |
| + return keys |
| +} |
| + |
| +// cleanupPrefix deletes every object AND version under the test prefix so |
| +// spike runs leave nothing behind in the bucket. |
| +func cleanupPrefix(t *testing.T, bucket, prefix, region string) { |
| + t.Helper() |
| + client := s3Client(t, region) |
| + ctx := context.Background() |
| + p := awss3.NewListObjectVersionsPaginator(client, &awss3.ListObjectVersionsInput{ |
| + Bucket: aws.String(bucket), |
| + Prefix: aws.String(prefix), |
| + }) |
| + for p.HasMorePages() { |
| + out, err := p.NextPage(ctx) |
| + if err != nil { |
| + t.Logf("cleanup list versions: %v", err) |
| + return |
| + } |
| + var objs []s3types.ObjectIdentifier |
| + for _, v := range out.Versions { |
| + objs = append(objs, s3types.ObjectIdentifier{Key: v.Key, VersionId: v.VersionId}) |
| + } |
| + for _, m := range out.DeleteMarkers { |
| + objs = append(objs, s3types.ObjectIdentifier{Key: m.Key, VersionId: m.VersionId}) |
| + } |
| + if len(objs) == 0 { |
| + continue |
| + } |
| + if _, err := client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{ |
| + Bucket: aws.String(bucket), |
| + Delete: &s3types.Delete{Objects: objs}, |
| + }); err != nil { |
| + t.Logf("cleanup delete: %v", err) |
| + } |
| + } |
| +} |
| + |
| +// TestHeldError_ErrorsAs is a unit test (no AWS): the package's contract is |
| +// that a contested lease surfaces as *HeldError through errors.As, carrying |
| +// the holder and expiry. |
| +func TestHeldError_ErrorsAs(t *testing.T) { |
| + exp := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) |
| + wrapped := fmt.Errorf("acquire lease: %w", &HeldError{Owner: "peer-1", ExpiresAt: exp}) |
| + var held *HeldError |
| + if !errors.As(wrapped, &held) { |
| + t.Fatalf("errors.As failed to find *HeldError in %v", wrapped) |
| + } |
| + if held.Owner != "peer-1" || !held.ExpiresAt.Equal(exp) { |
| + t.Fatalf("HeldError fields lost in wrap: %+v", held) |
| + } |
| + if msg := held.Error(); !strings.Contains(msg, "peer-1") { |
| + t.Fatalf("Error() should name the holder: %q", msg) |
| + } |
| +} |
| + |
| +// TestReplicateWithLease_ReplicatesToS3 proves Q1+Q2: the embedded library |
| +// (pure Go, modernc driver) replicates a real database to S3 under a lease. |
| +func TestReplicateWithLease_ReplicatesToS3(t *testing.T) { |
| + bucket, prefix, region := liveConfig(t) |
| + ctx := context.Background() |
| + |
| + dbPath := filepath.Join(t.TempDir(), "spike.db") |
| + seedDB(t, dbPath) |
| + |
| + r, err := ReplicateWithLease(ctx, dbPath, bucket, prefix, region) |
| + if err != nil { |
| + t.Fatalf("ReplicateWithLease: %v", err) |
| + } |
| + defer r.Close(ctx) |
| + |
| + if err := r.SyncAndWait(ctx); err != nil { |
| + t.Fatalf("SyncAndWait: %v", err) |
| + } |
| + |
| + keys := listKeys(t, bucket, prefix, region) |
| + t.Logf("objects under %s: %v", prefix, keys) |
| + var haveLock, haveLTX bool |
| + for _, k := range keys { |
| + if strings.HasSuffix(k, "lock.json") { |
| + haveLock = true |
| + } |
| + if strings.Contains(k, ".ltx") { |
| + haveLTX = true |
| + } |
| + } |
| + if !haveLock { |
| + t.Errorf("expected %s/lock.json in bucket, keys=%v", prefix, keys) |
| + } |
| + if !haveLTX { |
| + t.Errorf("expected replicated ltx data under %s, keys=%v", prefix, keys) |
| + } |
| + |
| + if err := r.Close(ctx); err != nil { |
| + t.Fatalf("Close: %v", err) |
| + } |
| +} |
| + |
| +// TestReplicateWithLease_MutualExclusion proves Q3: a second claimant on the |
| +// same replica path fails LOUDLY at lease acquisition with a typed error — |
| +// unlike the CLI, which silently absorbs the second writer. |
| +func TestReplicateWithLease_MutualExclusion(t *testing.T) { |
| + bucket, prefix, region := liveConfig(t) |
| + ctx := context.Background() |
| + |
| + dir := t.TempDir() |
| + db1 := filepath.Join(dir, "one.db") |
| + db2 := filepath.Join(dir, "two.db") |
| + seedDB(t, db1) |
| + seedDB(t, db2) |
| + |
| + first, err := ReplicateWithLease(ctx, db1, bucket, prefix, region) |
| + if err != nil { |
| + t.Fatalf("first claimant: %v", err) |
| + } |
| + defer first.Close(ctx) |
| + |
| + // Second claimant, same bucket+prefix: must fail, and fail typed. |
| + second, err := ReplicateWithLease(ctx, db2, bucket, prefix, region) |
| + if err == nil { |
| + second.Close(ctx) |
| + t.Fatal("second claimant acquired the lease while the first held it — mutual exclusion broken") |
| + } |
| + var lexists *HeldError |
| + if !errors.As(err, &lexists) { |
| + t.Fatalf("second claimant failed, but not with *HeldError: %v", err) |
| + } |
| + t.Logf("second claimant failed loudly (typed): %v", err) |
| + t.Logf(" holder=%q expires=%s", lexists.Owner, lexists.ExpiresAt.Format(time.RFC3339)) |
| + |
| + // Release, then the next claimant must succeed immediately (no TTL wait). |
| + if err := first.Close(ctx); err != nil { |
| + t.Fatalf("release: %v", err) |
| + } |
| + third, err := ReplicateWithLease(ctx, db2, bucket, prefix, region) |
| + if err != nil { |
| + t.Fatalf("claimant after release should succeed: %v", err) |
| + } |
| + if err := third.Close(ctx); err != nil { |
| + t.Fatalf("third close: %v", err) |
| + } |
| +} |