| 1 | // Ported from lchat internal/activator/replica_test.go. |
| 2 | // |
| 3 | // SPIKE tests (issue #550 Part 4 v2). Live S3 tests are gated on |
| 4 | // CARLOS_TEST_S3_BUCKET; without it everything skips so `go test ./...` |
| 5 | // stays green on CI and dev machines with no AWS credentials. |
| 6 | package lease |
| 7 | |
| 8 | import ( |
| 9 | "context" |
| 10 | "crypto/rand" |
| 11 | "database/sql" |
| 12 | "encoding/hex" |
| 13 | "errors" |
| 14 | "fmt" |
| 15 | "os" |
| 16 | "path/filepath" |
| 17 | "strings" |
| 18 | "testing" |
| 19 | "time" |
| 20 | |
| 21 | "github.com/aws/aws-sdk-go-v2/aws" |
| 22 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 23 | s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" |
| 24 | |
| 25 | _ "modernc.org/sqlite" |
| 26 | ) |
| 27 | |
| 28 | func liveConfig(t *testing.T) (bucket, prefix, region string) { |
| 29 | t.Helper() |
| 30 | bucket = os.Getenv("CARLOS_TEST_S3_BUCKET") |
| 31 | if bucket == "" { |
| 32 | t.Skip("CARLOS_TEST_S3_BUCKET not set; skipping live S3 spike test") |
| 33 | } |
| 34 | region = os.Getenv("CARLOS_TEST_S3_REGION") |
| 35 | if region == "" { |
| 36 | region = "eu-west-1" |
| 37 | } |
| 38 | b := make([]byte, 4) |
| 39 | _, _ = rand.Read(b) |
| 40 | prefix = fmt.Sprintf("spike/%d-%s", time.Now().Unix(), hex.EncodeToString(b)) |
| 41 | t.Cleanup(func() { cleanupPrefix(t, bucket, prefix, region) }) |
| 42 | return bucket, prefix, region |
| 43 | } |
| 44 | |
| 45 | // seedDB creates a WAL-mode SQLite database via modernc.org/sqlite (the |
| 46 | // driver this repo already uses — no cgo anywhere in the test). |
| 47 | func seedDB(t *testing.T, path string) { |
| 48 | t.Helper() |
| 49 | db, err := sql.Open("sqlite", path) |
| 50 | if err != nil { |
| 51 | t.Fatalf("open sqlite: %v", err) |
| 52 | } |
| 53 | defer db.Close() |
| 54 | if _, err := db.Exec(`PRAGMA journal_mode=WAL`); err != nil { |
| 55 | t.Fatalf("wal mode: %v", err) |
| 56 | } |
| 57 | if _, err := db.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil { |
| 58 | t.Fatalf("create table: %v", err) |
| 59 | } |
| 60 | for i := 0; i < 50; i++ { |
| 61 | if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-%d", i)); err != nil { |
| 62 | t.Fatalf("insert: %v", err) |
| 63 | } |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | // s3Client is the verification client the live tests list and clean up with. |
| 68 | // It deliberately goes through the package's OWN constructor rather than |
| 69 | // building its own: a helper that dialled stock AWS while the code under test |
| 70 | // honoured CARLOS_S3_ENDPOINT would fail every assertion against a compatible |
| 71 | // store and strand the objects it was meant to delete. |
| 72 | func s3Client(t *testing.T, region string) *awss3.Client { |
| 73 | t.Helper() |
| 74 | client, _, err := newLeaserClient(context.Background(), region) |
| 75 | if err != nil { |
| 76 | t.Fatalf("aws config: %v", err) |
| 77 | } |
| 78 | return client |
| 79 | } |
| 80 | |
| 81 | func listKeys(t *testing.T, bucket, prefix, region string) []string { |
| 82 | t.Helper() |
| 83 | client := s3Client(t, region) |
| 84 | var keys []string |
| 85 | p := awss3.NewListObjectsV2Paginator(client, &awss3.ListObjectsV2Input{ |
| 86 | Bucket: aws.String(bucket), |
| 87 | Prefix: aws.String(prefix), |
| 88 | }) |
| 89 | for p.HasMorePages() { |
| 90 | out, err := p.NextPage(context.Background()) |
| 91 | if err != nil { |
| 92 | t.Fatalf("list objects: %v", err) |
| 93 | } |
| 94 | for _, o := range out.Contents { |
| 95 | keys = append(keys, *o.Key) |
| 96 | } |
| 97 | } |
| 98 | return keys |
| 99 | } |
| 100 | |
| 101 | // cleanupPrefix deletes every object AND version under the test prefix so |
| 102 | // spike runs leave nothing behind in the bucket. |
| 103 | func cleanupPrefix(t *testing.T, bucket, prefix, region string) { |
| 104 | t.Helper() |
| 105 | client := s3Client(t, region) |
| 106 | ctx := context.Background() |
| 107 | p := awss3.NewListObjectVersionsPaginator(client, &awss3.ListObjectVersionsInput{ |
| 108 | Bucket: aws.String(bucket), |
| 109 | Prefix: aws.String(prefix), |
| 110 | }) |
| 111 | for p.HasMorePages() { |
| 112 | out, err := p.NextPage(ctx) |
| 113 | if err != nil { |
| 114 | t.Logf("cleanup list versions: %v", err) |
| 115 | return |
| 116 | } |
| 117 | var objs []s3types.ObjectIdentifier |
| 118 | for _, v := range out.Versions { |
| 119 | objs = append(objs, s3types.ObjectIdentifier{Key: v.Key, VersionId: v.VersionId}) |
| 120 | } |
| 121 | for _, m := range out.DeleteMarkers { |
| 122 | objs = append(objs, s3types.ObjectIdentifier{Key: m.Key, VersionId: m.VersionId}) |
| 123 | } |
| 124 | if len(objs) == 0 { |
| 125 | continue |
| 126 | } |
| 127 | if _, err := client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{ |
| 128 | Bucket: aws.String(bucket), |
| 129 | Delete: &s3types.Delete{Objects: objs}, |
| 130 | }); err != nil { |
| 131 | t.Logf("cleanup delete: %v", err) |
| 132 | } |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | // TestHeldError_ErrorsAs is a unit test (no AWS): the package's contract is |
| 137 | // that a contested lease surfaces as *HeldError through errors.As, carrying |
| 138 | // the holder and expiry. |
| 139 | func TestHeldError_ErrorsAs(t *testing.T) { |
| 140 | exp := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) |
| 141 | wrapped := fmt.Errorf("acquire lease: %w", &HeldError{Owner: "peer-1", ExpiresAt: exp}) |
| 142 | var held *HeldError |
| 143 | if !errors.As(wrapped, &held) { |
| 144 | t.Fatalf("errors.As failed to find *HeldError in %v", wrapped) |
| 145 | } |
| 146 | if held.Owner != "peer-1" || !held.ExpiresAt.Equal(exp) { |
| 147 | t.Fatalf("HeldError fields lost in wrap: %+v", held) |
| 148 | } |
| 149 | if msg := held.Error(); !strings.Contains(msg, "peer-1") { |
| 150 | t.Fatalf("Error() should name the holder: %q", msg) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // TestReplicateWithLease_ReplicatesToS3 proves Q1+Q2: the embedded library |
| 155 | // (pure Go, modernc driver) replicates a real database to S3 under a lease. |
| 156 | func TestReplicateWithLease_ReplicatesToS3(t *testing.T) { |
| 157 | bucket, prefix, region := liveConfig(t) |
| 158 | ctx := context.Background() |
| 159 | |
| 160 | dbPath := filepath.Join(t.TempDir(), "spike.db") |
| 161 | seedDB(t, dbPath) |
| 162 | |
| 163 | r, err := ReplicateWithLease(ctx, dbPath, bucket, prefix, region) |
| 164 | if err != nil { |
| 165 | t.Fatalf("ReplicateWithLease: %v", err) |
| 166 | } |
| 167 | defer r.Close(ctx) |
| 168 | |
| 169 | if err := r.SyncAndWait(ctx); err != nil { |
| 170 | t.Fatalf("SyncAndWait: %v", err) |
| 171 | } |
| 172 | |
| 173 | keys := listKeys(t, bucket, prefix, region) |
| 174 | t.Logf("objects under %s: %v", prefix, keys) |
| 175 | var haveLock, haveLTX bool |
| 176 | for _, k := range keys { |
| 177 | if strings.HasSuffix(k, "lock.json") { |
| 178 | haveLock = true |
| 179 | } |
| 180 | if strings.Contains(k, ".ltx") { |
| 181 | haveLTX = true |
| 182 | } |
| 183 | } |
| 184 | if !haveLock { |
| 185 | t.Errorf("expected %s/lock.json in bucket, keys=%v", prefix, keys) |
| 186 | } |
| 187 | if !haveLTX { |
| 188 | t.Errorf("expected replicated ltx data under %s, keys=%v", prefix, keys) |
| 189 | } |
| 190 | |
| 191 | if err := r.Close(ctx); err != nil { |
| 192 | t.Fatalf("Close: %v", err) |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | // TestReplicateWithLease_MutualExclusion proves Q3: a second claimant on the |
| 197 | // same replica path fails LOUDLY at lease acquisition with a typed error — |
| 198 | // unlike the CLI, which silently absorbs the second writer. |
| 199 | func TestReplicateWithLease_MutualExclusion(t *testing.T) { |
| 200 | bucket, prefix, region := liveConfig(t) |
| 201 | ctx := context.Background() |
| 202 | |
| 203 | dir := t.TempDir() |
| 204 | db1 := filepath.Join(dir, "one.db") |
| 205 | db2 := filepath.Join(dir, "two.db") |
| 206 | seedDB(t, db1) |
| 207 | seedDB(t, db2) |
| 208 | |
| 209 | first, err := ReplicateWithLease(ctx, db1, bucket, prefix, region) |
| 210 | if err != nil { |
| 211 | t.Fatalf("first claimant: %v", err) |
| 212 | } |
| 213 | defer first.Close(ctx) |
| 214 | |
| 215 | // Second claimant, same bucket+prefix: must fail, and fail typed. |
| 216 | second, err := ReplicateWithLease(ctx, db2, bucket, prefix, region) |
| 217 | if err == nil { |
| 218 | second.Close(ctx) |
| 219 | t.Fatal("second claimant acquired the lease while the first held it — mutual exclusion broken") |
| 220 | } |
| 221 | var lexists *HeldError |
| 222 | if !errors.As(err, &lexists) { |
| 223 | t.Fatalf("second claimant failed, but not with *HeldError: %v", err) |
| 224 | } |
| 225 | t.Logf("second claimant failed loudly (typed): %v", err) |
| 226 | t.Logf(" holder=%q expires=%s", lexists.Owner, lexists.ExpiresAt.Format(time.RFC3339)) |
| 227 | |
| 228 | // Release, then the next claimant must succeed immediately (no TTL wait). |
| 229 | if err := first.Close(ctx); err != nil { |
| 230 | t.Fatalf("release: %v", err) |
| 231 | } |
| 232 | third, err := ReplicateWithLease(ctx, db2, bucket, prefix, region) |
| 233 | if err != nil { |
| 234 | t.Fatalf("claimant after release should succeed: %v", err) |
| 235 | } |
| 236 | if err := third.Close(ctx); err != nil { |
| 237 | t.Fatalf("third close: %v", err) |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | // TestDarkHostCannotReclaimAfterFailover proves the dark-then-returns half of |
| 242 | // failover (issue #53): host A holds the lease and goes DARK — stops renewing |
| 243 | // but never releases, the kill -9 / power-cut shape. Its lock.json stays in |
| 244 | // S3 with the expiry already stamped on it, so: |
| 245 | // |
| 246 | // 1. while that expiry is still in the future a peer is refused — dark is |
| 247 | // not the same as expired, and nobody may jump a live lease; |
| 248 | // 2. once it lapses, host B claims it — the failover wake; |
| 249 | // 3. A comes back and tries to pick up where it left off, and MUST lose, |
| 250 | // loudly, with a *HeldError. A returning dark host silently reclaiming a |
| 251 | // lease that has moved on is the split-brain this lock exists to prevent. |
| 252 | // |
| 253 | // The other half of failover — a fallback host's activator waking the |
| 254 | // instance from the regional bucket — is the existing activator wake path and |
| 255 | // is covered by the activator's own tests; nothing about it is new here. |
| 256 | // |
| 257 | // Runs on a 15s TTL through the AcquireTTL seam; at the production LeaseTTL |
| 258 | // the same sequence would take 15 minutes. |
| 259 | func TestDarkHostCannotReclaimAfterFailover(t *testing.T) { |
| 260 | bucket, prefix, region := liveConfig(t) |
| 261 | ctx := context.Background() |
| 262 | const shortTTL = 15 * time.Second |
| 263 | |
| 264 | a, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL) |
| 265 | if err != nil { |
| 266 | t.Fatalf("A initial acquire: %v", err) |
| 267 | } |
| 268 | acquiredAt := time.Now() |
| 269 | // A goes dark. rn.stop() cancels the renewal goroutine's context and |
| 270 | // waits for it to exit; it makes no S3 call of any kind, so lock.json |
| 271 | // survives untouched and simply runs out its stamped TTL. Without this |
| 272 | // line A would renew every shortTTL/3 and step 2 below could never |
| 273 | // happen — that is what makes the stop load-bearing. |
| 274 | a.rn.stop() |
| 275 | |
| 276 | // Dark is not expired: A's lock is still in force, so B must be refused — |
| 277 | // and refused BY THE LEASE, not by a throttle, a 5xx or a credential |
| 278 | // hiccup, which would let this leg pass while proving nothing. So the |
| 279 | // refusal has to be a *HeldError carrying a live expiry, exactly as |
| 280 | // TestReplicateWithLease_MutualExclusion demands of its second claimant. |
| 281 | // (Both "hosts" run in this one process and therefore share the leaser's |
| 282 | // owner string, so the expiry — not the name — is what identifies the |
| 283 | // holder here.) |
| 284 | peer, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL) |
| 285 | if err == nil { |
| 286 | elapsed := time.Since(acquiredAt) |
| 287 | _ = peer.Release(ctx) |
| 288 | if elapsed >= shortTTL { |
| 289 | t.Fatalf("second acquire only returned %s after A acquired, past the %s TTL: this environment is too slow for the test's timing margins, which is not the same as a mutual-exclusion failure", elapsed, shortTTL) |
| 290 | } |
| 291 | t.Fatalf("B acquired after %s while A's un-expired %s lease was still held", elapsed, shortTTL) |
| 292 | } |
| 293 | var stillHeld *HeldError |
| 294 | if !errors.As(err, &stillHeld) { |
| 295 | t.Fatalf("peer refused while A was dark-but-live, but not with *HeldError: %v — an S3 or credential failure here would make this leg vacuous", err) |
| 296 | } |
| 297 | if !stillHeld.ExpiresAt.After(time.Now()) { |
| 298 | t.Fatalf("peer refused by an ALREADY-EXPIRED lease (expires %s, %s after A acquired); want A's still-live lease", stillHeld.ExpiresAt.Format(time.RFC3339), time.Since(acquiredAt)) |
| 299 | } |
| 300 | t.Logf("dark is not expired: peer refused by A's live lease until %s", stillHeld.ExpiresAt.Format(time.RFC3339)) |
| 301 | |
| 302 | time.Sleep(shortTTL + 5*time.Second) // let A's abandoned lease lapse |
| 303 | |
| 304 | b, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL) |
| 305 | if err != nil { |
| 306 | t.Fatalf("B acquire after A's lease expired (the failover wake): %v", err) |
| 307 | } |
| 308 | bHeldFrom := time.Now() |
| 309 | defer b.Release(ctx) |
| 310 | |
| 311 | // A returns from the dead and tries to resume. |
| 312 | if revenant, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL); err == nil { |
| 313 | elapsed := time.Since(bHeldFrom) |
| 314 | _ = revenant.Release(ctx) |
| 315 | if elapsed >= shortTTL { |
| 316 | // B's own lease had lapsed by then (B is not renewed here either), |
| 317 | // so A winning proves nothing. Say so rather than cry split brain. |
| 318 | t.Fatalf("A's re-acquire only returned %s after B took over, past the %s TTL: this environment is too slow for the test's timing margins", elapsed, shortTTL) |
| 319 | } |
| 320 | t.Fatal("returning dark host reclaimed the lease B now holds — split brain") |
| 321 | } else { |
| 322 | var held *HeldError |
| 323 | if !errors.As(err, &held) { |
| 324 | t.Fatalf("returning dark host got %v, want *HeldError (it must lose the race)", err) |
| 325 | } |
| 326 | // The refusal has to come from B's LIVE lease. An expiry already in |
| 327 | // the past would mean the acquire lost for some other reason and the |
| 328 | // test would be proving nothing. |
| 329 | if !held.ExpiresAt.After(time.Now()) { |
| 330 | t.Fatalf("returning host refused by an already-expired lease (expires %s); want B's live lease", held.ExpiresAt.Format(time.RFC3339)) |
| 331 | } |
| 332 | t.Logf("returning dark host lost the race, as required: %v", err) |
| 333 | } |
| 334 | } |
| 335 | |
amadan