| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "database/sql" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "net/http" |
| 12 | "net/http/httptest" |
| 13 | "path/filepath" |
| 14 | "sync" |
| 15 | "testing" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/aws/aws-sdk-go-v2/aws" |
| 19 | s3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 20 | "github.com/aws/smithy-go" |
| 21 | "github.com/benbjohnson/litestream" |
| 22 | lsfile "github.com/benbjohnson/litestream/file" |
| 23 | lss3 "github.com/benbjohnson/litestream/s3" |
| 24 | "github.com/superfly/ltx" |
| 25 | ) |
| 26 | |
| 27 | type lockStore struct { |
| 28 | mu sync.Mutex |
| 29 | body []byte |
| 30 | etag string |
| 31 | puts, deletes int |
| 32 | beforePut func(context.Context, *s3.PutObjectInput) error |
| 33 | } |
| 34 | |
| 35 | func (s *lockStore) GetObject(ctx context.Context, p *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) { |
| 36 | s.mu.Lock() |
| 37 | defer s.mu.Unlock() |
| 38 | if s.body == nil { |
| 39 | return nil, &smithy.GenericAPIError{Code: "NoSuchKey"} |
| 40 | } |
| 41 | return &s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader(bytes.Clone(s.body))), ETag: aws.String(s.etag)}, nil |
| 42 | } |
| 43 | func (s *lockStore) PutObject(ctx context.Context, p *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) { |
| 44 | raw, e := io.ReadAll(p.Body) |
| 45 | if e != nil { |
| 46 | return nil, e |
| 47 | } |
| 48 | if s.beforePut != nil { |
| 49 | if e := s.beforePut(ctx, p); e != nil { |
| 50 | return nil, e |
| 51 | } |
| 52 | } |
| 53 | s.mu.Lock() |
| 54 | defer s.mu.Unlock() |
| 55 | if (aws.ToString(p.IfNoneMatch) == "*" && s.body != nil) || (p.IfMatch != nil && aws.ToString(p.IfMatch) != s.etag) { |
| 56 | return nil, &smithy.GenericAPIError{Code: "PreconditionFailed"} |
| 57 | } |
| 58 | s.puts++ |
| 59 | s.etag = fmt.Sprintf("\"etag-%d\"", s.puts) |
| 60 | s.body = raw |
| 61 | return &s3.PutObjectOutput{ETag: aws.String(s.etag)}, nil |
| 62 | } |
| 63 | func (s *lockStore) DeleteObject(ctx context.Context, p *s3.DeleteObjectInput, _ ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { |
| 64 | s.mu.Lock() |
| 65 | defer s.mu.Unlock() |
| 66 | if p.IfMatch == nil || aws.ToString(p.IfMatch) != s.etag { |
| 67 | return nil, &smithy.GenericAPIError{Code: "PreconditionFailed"} |
| 68 | } |
| 69 | s.deletes++ |
| 70 | s.body = nil |
| 71 | s.etag = "" |
| 72 | return &s3.DeleteObjectOutput{}, nil |
| 73 | } |
| 74 | func (s *lockStore) counts() (int, int) { s.mu.Lock(); defer s.mu.Unlock(); return s.puts, s.deletes } |
| 75 | func (s *lockStore) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 76 | var err error |
| 77 | switch r.Method { |
| 78 | case "GET": |
| 79 | var out *s3.GetObjectOutput |
| 80 | out, err = s.GetObject(r.Context(), &s3.GetObjectInput{}) |
| 81 | if err == nil { |
| 82 | w.Header().Set("ETag", aws.ToString(out.ETag)) |
| 83 | io.Copy(w, out.Body) |
| 84 | return |
| 85 | } |
| 86 | case "PUT": |
| 87 | p := &s3.PutObjectInput{Body: r.Body} |
| 88 | if v := r.Header.Get("If-Match"); v != "" { |
| 89 | p.IfMatch = aws.String(v) |
| 90 | } |
| 91 | if v := r.Header.Get("If-None-Match"); v != "" { |
| 92 | p.IfNoneMatch = aws.String(v) |
| 93 | } |
| 94 | var out *s3.PutObjectOutput |
| 95 | out, err = s.PutObject(r.Context(), p) |
| 96 | if err == nil { |
| 97 | w.Header().Set("ETag", aws.ToString(out.ETag)) |
| 98 | return |
| 99 | } |
| 100 | case "DELETE": |
| 101 | _, err = s.DeleteObject(r.Context(), &s3.DeleteObjectInput{IfMatch: aws.String(r.Header.Get("If-Match"))}) |
| 102 | if err == nil { |
| 103 | w.WriteHeader(204) |
| 104 | return |
| 105 | } |
| 106 | } |
| 107 | var api *smithy.GenericAPIError |
| 108 | if errors.As(err, &api) { |
| 109 | code := 412 |
| 110 | if api.Code == "NoSuchKey" { |
| 111 | code = 404 |
| 112 | } |
| 113 | w.Header().Set("Content-Type", "application/xml") |
| 114 | w.WriteHeader(code) |
| 115 | fmt.Fprintf(w, "<Error><Code>%s</Code></Error>", api.Code) |
| 116 | return |
| 117 | } |
| 118 | w.WriteHeader(500) |
| 119 | } |
| 120 | func localHeld(t *testing.T, backend *lockStore, ttl time.Duration) *Held { |
| 121 | t.Helper() |
| 122 | srv := httptest.NewServer(backend) |
| 123 | t.Cleanup(srv.Close) |
| 124 | fakeCreds(t) |
| 125 | t.Setenv("CARLOS_S3_ENDPOINT", srv.URL) |
| 126 | h, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl) |
| 127 | if e != nil { |
| 128 | t.Fatal(e) |
| 129 | } |
| 130 | t.Cleanup(func() { _ = h.Release(context.Background()) }) |
| 131 | return h |
| 132 | } |
| 133 | func await(t *testing.T, ch <-chan struct{}) { |
| 134 | t.Helper() |
| 135 | select { |
| 136 | case <-ch: |
| 137 | case <-time.After(5 * time.Second): |
| 138 | t.Fatal("timed out") |
| 139 | } |
| 140 | } |
| 141 | func restoreCount(t *testing.T, replica, out string) int { |
| 142 | t.Helper() |
| 143 | r := litestream.NewReplicaWithClient(nil, lsfile.NewReplicaClient(replica)) |
| 144 | opt := litestream.NewRestoreOptions() |
| 145 | opt.OutputPath = out |
| 146 | if e := r.Restore(context.Background(), opt); e != nil { |
| 147 | t.Fatal(e) |
| 148 | } |
| 149 | d, e := sql.Open("sqlite", "file:"+out+"?mode=ro") |
| 150 | if e != nil { |
| 151 | t.Fatal(e) |
| 152 | } |
| 153 | defer d.Close() |
| 154 | var n int |
| 155 | var integrity string |
| 156 | if e = d.QueryRow("PRAGMA integrity_check").Scan(&integrity); e != nil || integrity != "ok" { |
| 157 | t.Fatalf("integrity %q %v", integrity, e) |
| 158 | } |
| 159 | if e = d.QueryRow("SELECT count(*) FROM t").Scan(&n); e != nil { |
| 160 | t.Fatal(e) |
| 161 | } |
| 162 | var contents string |
| 163 | if e = d.QueryRow("SELECT group_concat(body, '|') FROM (SELECT body FROM t ORDER BY id)").Scan(&contents); e != nil { |
| 164 | t.Fatal(e) |
| 165 | } |
| 166 | if n == 5 && contents != "post-vacuum|post-vacuum|post-vacuum|post-vacuum|post-vacuum" { |
| 167 | t.Fatalf("restored content receipt differs: %q", contents) |
| 168 | } |
| 169 | return n |
| 170 | } |
| 171 | func TestHeldCheckpointVacuumRestoreAndTransfer(t *testing.T) { |
| 172 | backend := &lockStore{} |
| 173 | h := localHeld(t, backend, 10*time.Second) |
| 174 | path := filepath.Join(t.TempDir(), "source.db") |
| 175 | seedDB(t, path) |
| 176 | replica := t.TempDir() |
| 177 | h.replicaClient = lsfile.NewReplicaClient(replica) |
| 178 | rn := h.rn |
| 179 | receipt, e := h.Checkpoint(context.Background(), path) |
| 180 | if e != nil { |
| 181 | t.Fatal(e) |
| 182 | } |
| 183 | if receipt.CompletedAt.IsZero() || !receipt.CompletedAt.Before(receipt.SafeUntil) || h.rn != rn { |
| 184 | t.Fatal("invalid checkpoint receipt/ownership") |
| 185 | } |
| 186 | if n := restoreCount(t, replica, filepath.Join(t.TempDir(), "old.db")); n != 50 { |
| 187 | t.Fatal(n) |
| 188 | } |
| 189 | d, e := sql.Open("sqlite", path) |
| 190 | if e != nil { |
| 191 | t.Fatal(e) |
| 192 | } |
| 193 | _, e = d.Exec("DELETE FROM t WHERE id>5; UPDATE t SET body='post-vacuum'; PRAGMA wal_checkpoint(TRUNCATE); VACUUM; PRAGMA wal_checkpoint(TRUNCATE)") |
| 194 | if e != nil { |
| 195 | t.Fatal(e) |
| 196 | } |
| 197 | if e = d.Close(); e != nil { |
| 198 | t.Fatal(e) |
| 199 | } |
| 200 | if _, e = h.Checkpoint(context.Background(), path); e != nil { |
| 201 | t.Fatal(e) |
| 202 | } |
| 203 | if n := restoreCount(t, replica, filepath.Join(t.TempDir(), "new.db")); n != 5 { |
| 204 | t.Fatalf("stale replica: %d", n) |
| 205 | } |
| 206 | if _, deletes := backend.counts(); deletes != 0 { |
| 207 | t.Fatal("checkpoint released lock") |
| 208 | } |
| 209 | // The actual leaser sees the same still-owned lock after both checkpoints. |
| 210 | if peer, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 10*time.Second); e == nil { |
| 211 | peer.Release(context.Background()) |
| 212 | t.Fatal("acquisition gap") |
| 213 | } |
| 214 | r, e := h.Replicate(context.Background(), path, 0) |
| 215 | if e != nil { |
| 216 | t.Fatal(e) |
| 217 | } |
| 218 | defer r.Close(context.Background()) |
| 219 | if r.Lost() != h.Lost() { |
| 220 | t.Fatal("transfer replaced loss channel") |
| 221 | } |
| 222 | select { |
| 223 | case <-r.Lost(): |
| 224 | t.Fatal("healthy transfer signaled loss") |
| 225 | default: |
| 226 | } |
| 227 | if r.rn != rn { |
| 228 | t.Fatal("renewer replaced") |
| 229 | } |
| 230 | if e = h.Release(context.Background()); !errors.Is(e, ErrSpent) { |
| 231 | t.Fatalf("spent release %v", e) |
| 232 | } |
| 233 | if _, e = h.SafeUntil(); !errors.Is(e, ErrSpent) { |
| 234 | t.Fatal("spent deadline accepted") |
| 235 | } |
| 236 | if _, e = r.SafeUntil(); e != nil { |
| 237 | t.Fatal(e) |
| 238 | } |
| 239 | if _, deletes := backend.counts(); deletes != 0 { |
| 240 | t.Fatal("spent Held released successor") |
| 241 | } |
| 242 | if e = r.Close(context.Background()); e != nil { |
| 243 | t.Fatal(e) |
| 244 | } |
| 245 | if _, e = r.SafeUntil(); e == nil { |
| 246 | t.Fatal("stopped deadline accepted") |
| 247 | } |
| 248 | select { |
| 249 | case <-r.Lost(): |
| 250 | t.Fatal("normal Close signaled loss") |
| 251 | default: |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | type blockedReplica struct { |
| 256 | litestream.ReplicaClient |
| 257 | entered chan struct{} |
| 258 | release chan struct{} |
| 259 | once sync.Once |
| 260 | fail error |
| 261 | } |
| 262 | |
| 263 | func (b *blockedReplica) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) { |
| 264 | b.once.Do(func() { close(b.entered) }) |
| 265 | select { |
| 266 | case <-b.release: |
| 267 | case <-ctx.Done(): |
| 268 | return nil, ctx.Err() |
| 269 | } |
| 270 | if b.fail != nil { |
| 271 | return nil, b.fail |
| 272 | } |
| 273 | return b.ReplicaClient.WriteLTXFile(ctx, level, min, max, r) |
| 274 | } |
| 275 | func TestCheckpointExclusionAndPoison(t *testing.T) { |
| 276 | for _, fail := range []bool{false, true} { |
| 277 | t.Run(fmt.Sprint(fail), func(t *testing.T) { |
| 278 | backend := &lockStore{} |
| 279 | h := localHeld(t, backend, 10*time.Second) |
| 280 | path := filepath.Join(t.TempDir(), "source.db") |
| 281 | seedDB(t, path) |
| 282 | b := &blockedReplica{ReplicaClient: lsfile.NewReplicaClient(t.TempDir()), entered: make(chan struct{}), release: make(chan struct{})} |
| 283 | if fail { |
| 284 | b.fail = errors.New("synthetic replica write refusal") |
| 285 | } |
| 286 | h.replicaClient = b |
| 287 | opCtx, cancel := context.WithTimeout(context.Background(), time.Second) |
| 288 | defer cancel() |
| 289 | done := make(chan error, 1) |
| 290 | go func() { _, e := h.Checkpoint(opCtx, path); done <- e }() |
| 291 | await(t, b.entered) |
| 292 | if _, e := h.Checkpoint(context.Background(), path); !errors.Is(e, ErrBusy) { |
| 293 | t.Fatal(e) |
| 294 | } |
| 295 | if _, e := h.Replicate(context.Background(), path, 0); !errors.Is(e, ErrBusy) { |
| 296 | t.Fatal(e) |
| 297 | } |
| 298 | if e := h.Release(context.Background()); !errors.Is(e, ErrBusy) { |
| 299 | t.Fatal(e) |
| 300 | } |
| 301 | close(b.release) |
| 302 | e := <-done |
| 303 | if fail { |
| 304 | if e == nil { |
| 305 | t.Fatal("failed sync accepted") |
| 306 | } |
| 307 | if _, e = h.Checkpoint(context.Background(), path); e == nil { |
| 308 | t.Fatal("poisoned checkpoint") |
| 309 | } |
| 310 | if _, e = h.Replicate(context.Background(), path, 0); e == nil { |
| 311 | t.Fatal("poisoned replicate") |
| 312 | } |
| 313 | if _, e = h.SafeUntil(); e == nil { |
| 314 | t.Fatal("poisoned command deadline") |
| 315 | } |
| 316 | } else if e != nil { |
| 317 | t.Fatal(e) |
| 318 | } |
| 319 | if e = h.rn.Err(); e != nil { |
| 320 | t.Fatalf("ownership released on checkpoint outcome: %v", e) |
| 321 | } |
| 322 | if _, deletes := backend.counts(); deletes != 0 { |
| 323 | t.Fatal("unexpected release") |
| 324 | } |
| 325 | if e = h.Release(context.Background()); e != nil { |
| 326 | t.Fatal(e) |
| 327 | } |
| 328 | }) |
| 329 | } |
| 330 | } |
| 331 | |
| 332 | func TestBlockedRenewalDeadlinePrecedesTakeover(t *testing.T) { |
| 333 | entered := make(chan struct{}) |
| 334 | var once sync.Once |
| 335 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 336 | if p.IfMatch != nil { |
| 337 | once.Do(func() { close(entered) }) |
| 338 | <-ctx.Done() |
| 339 | return ctx.Err() |
| 340 | } |
| 341 | return nil |
| 342 | }} |
| 343 | // marginForTTL gives short test TTLs a margin of ttl/10. At 900ms that |
| 344 | // was 90ms, less than a loaded CI runner's scheduling delay (#268); 3s |
| 345 | // gives 300ms and checks the same property. Production's margin is 5s. |
| 346 | ttl := 3 * time.Second |
| 347 | h := localHeld(t, backend, ttl) |
| 348 | until, e := h.SafeUntil() |
| 349 | if e != nil { |
| 350 | t.Fatal(e) |
| 351 | } |
| 352 | await(t, entered) |
| 353 | // A worker observes cancellation without waiting for the blocked remote call. |
| 354 | await(t, h.Lost()) |
| 355 | if e = h.Err(); e == nil { |
| 356 | t.Fatal("blocked renewal remained healthy") |
| 357 | } |
| 358 | if time.Since(until) > ttl/10 { |
| 359 | t.Fatal("loss notification consumed termination margin") |
| 360 | } |
| 361 | backend.mu.Lock() |
| 362 | var original litestream.Lease |
| 363 | json.Unmarshal(backend.body, &original) |
| 364 | backend.mu.Unlock() |
| 365 | if !time.Now().Before(original.ExpiresAt) { |
| 366 | t.Fatal("termination did not precede lease expiry") |
| 367 | } |
| 368 | if peer, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl); e == nil { |
| 369 | peer.Release(context.Background()) |
| 370 | t.Fatal("peer took live lock") |
| 371 | } |
| 372 | time.Sleep(time.Until(original.ExpiresAt) + 20*time.Millisecond) |
| 373 | // A separate real leaser can take over the expired CAS object. Its acquisition |
| 374 | // uses If-Match, so disable the synthetic blocked-renewal hook first. |
| 375 | backend.beforePut = nil |
| 376 | peer, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl) |
| 377 | if e != nil { |
| 378 | t.Fatal(e) |
| 379 | } |
| 380 | defer peer.Release(context.Background()) |
| 381 | if e = h.Release(context.Background()); e == nil { |
| 382 | t.Fatal("stale release erased peer") |
| 383 | } |
| 384 | if e = peer.Err(); e != nil { |
| 385 | t.Fatal(e) |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | func TestLateRenewalCannotResurrect(t *testing.T) { |
| 390 | gate := make(chan struct{}) |
| 391 | entered := make(chan struct{}) |
| 392 | var once sync.Once |
| 393 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 394 | if p.IfMatch != nil { |
| 395 | once.Do(func() { close(entered) }) |
| 396 | <-gate |
| 397 | } |
| 398 | return nil |
| 399 | }} |
| 400 | leaser := lss3.NewLeaser() |
| 401 | leaser.SetClient(backend) |
| 402 | leaser.TTL = 300 * time.Millisecond |
| 403 | started := time.Now() |
| 404 | lock, e := leaser.AcquireLease(context.Background()) |
| 405 | if e != nil { |
| 406 | t.Fatal(e) |
| 407 | } |
| 408 | rn := startRenewer(leaser, lock, started, 30*time.Millisecond) |
| 409 | await(t, entered) |
| 410 | await(t, rn.ctx.Done()) |
| 411 | if !errors.Is(rn.Err(), ErrLost) { |
| 412 | t.Fatal("deadline not latched") |
| 413 | } |
| 414 | close(gate) |
| 415 | await(t, rn.done) |
| 416 | if !errors.Is(rn.Err(), ErrLost) { |
| 417 | t.Fatal("late acknowledgement resurrected ownership") |
| 418 | } |
| 419 | if _, e = rn.SafeUntil(); e == nil { |
| 420 | t.Fatal("late acknowledgement extended deadline") |
| 421 | } |
| 422 | leaser.ReleaseLease(context.Background(), rn.stop()) |
| 423 | } |
| 424 | |
| 425 | func TestAcquireDeadlineStartsBeforeRemoteRequest(t *testing.T) { |
| 426 | entered := make(chan time.Time, 1) |
| 427 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 428 | if p.IfMatch == nil { |
| 429 | entered <- time.Now() |
| 430 | time.Sleep(120 * time.Millisecond) |
| 431 | } |
| 432 | return nil |
| 433 | }} |
| 434 | ttl := time.Second |
| 435 | h := localHeld(t, backend, ttl) |
| 436 | beforeAck := <-entered |
| 437 | until, e := h.SafeUntil() |
| 438 | if e != nil { |
| 439 | t.Fatal(e) |
| 440 | } |
| 441 | if until.After(beforeAck.Add(ttl - marginForTTL(ttl))) { |
| 442 | t.Fatal("acquisition latency extended safe deadline") |
| 443 | } |
| 444 | if until == until.Round(0) { |
| 445 | t.Fatal("deadline lost monotonic clock") |
| 446 | } |
| 447 | } |
| 448 | |
| 449 | func TestAcquireLateSuccessGrantsNoHandle(t *testing.T) { |
| 450 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 451 | if p.IfMatch == nil { |
| 452 | time.Sleep(350 * time.Millisecond) |
| 453 | } |
| 454 | return nil |
| 455 | }} |
| 456 | srv := httptest.NewServer(backend) |
| 457 | defer srv.Close() |
| 458 | fakeCreds(t) |
| 459 | t.Setenv("CARLOS_S3_ENDPOINT", srv.URL) |
| 460 | h, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 300*time.Millisecond) |
| 461 | if e == nil || h != nil { |
| 462 | if h != nil { |
| 463 | h.Release(context.Background()) |
| 464 | } |
| 465 | t.Fatal("late acquisition granted authority") |
| 466 | } |
| 467 | if _, deletes := backend.counts(); deletes != 1 { |
| 468 | t.Fatal("known late acquisition was not conditionally cleaned up") |
| 469 | } |
| 470 | } |
| 471 | |
| 472 | func TestCheckpointLossCancelsBlockedSync(t *testing.T) { |
| 473 | path := filepath.Join(t.TempDir(), "source.db") |
| 474 | seedDB(t, path) |
| 475 | b := &blockedReplica{ReplicaClient: lsfile.NewReplicaClient(t.TempDir()), entered: make(chan struct{}), release: make(chan struct{})} |
| 476 | entered := make(chan struct{}) |
| 477 | var once sync.Once |
| 478 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 479 | if p.IfMatch != nil { |
| 480 | // Keep renewal healthy until the upload is actually blocked. |
| 481 | // Setup and filesystem latency must not consume the test lease. |
| 482 | select { |
| 483 | case <-b.entered: |
| 484 | default: |
| 485 | return nil |
| 486 | } |
| 487 | once.Do(func() { close(entered) }) |
| 488 | <-ctx.Done() |
| 489 | return ctx.Err() |
| 490 | } |
| 491 | return nil |
| 492 | }} |
| 493 | h := localHeld(t, backend, 600*time.Millisecond) |
| 494 | h.replicaClient = b |
| 495 | done := make(chan error, 1) |
| 496 | go func() { _, e := h.Checkpoint(context.Background(), path); done <- e }() |
| 497 | await(t, b.entered) |
| 498 | await(t, entered) |
| 499 | select { |
| 500 | case e := <-done: |
| 501 | if e == nil { |
| 502 | t.Fatal("lost lease returned receipt") |
| 503 | } |
| 504 | case <-time.After(2 * time.Second): |
| 505 | t.Fatal("loss did not cancel checkpoint") |
| 506 | } |
| 507 | if _, deletes := backend.counts(); deletes != 0 { |
| 508 | t.Fatal("checkpoint released after loss") |
| 509 | } |
| 510 | if _, e := h.SafeUntil(); e == nil { |
| 511 | t.Fatal("failed checkpoint usable") |
| 512 | } |
| 513 | } |
| 514 | |
| 515 | func TestCheckpointOpenFailureLeavesOwnedPoisonedHandle(t *testing.T) { |
| 516 | backend := &lockStore{} |
| 517 | h := localHeld(t, backend, 10*time.Second) |
| 518 | // A directory is never a SQLite database; the real store Open must refuse. |
| 519 | if receipt, e := h.Checkpoint(context.Background(), t.TempDir()); e == nil || !receipt.CompletedAt.IsZero() { |
| 520 | t.Fatal("failed open returned receipt") |
| 521 | } |
| 522 | if e := h.rn.Err(); e != nil { |
| 523 | t.Fatal("open failure lost held renewal") |
| 524 | } |
| 525 | if _, e := h.Replicate(context.Background(), "unused", 0); e == nil { |
| 526 | t.Fatal("open failure did not poison") |
| 527 | } |
| 528 | if _, deletes := backend.counts(); deletes != 0 { |
| 529 | t.Fatal("open failure released lease") |
| 530 | } |
| 531 | } |
| 532 | |
| 533 | func TestRenewDeadlineStartsBeforeRemoteRequest(t *testing.T) { |
| 534 | entered := make(chan time.Time, 1) |
| 535 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 536 | if p.IfMatch != nil { |
| 537 | select { |
| 538 | case entered <- time.Now(): |
| 539 | default: |
| 540 | } |
| 541 | time.Sleep(100 * time.Millisecond) |
| 542 | } |
| 543 | return nil |
| 544 | }} |
| 545 | leaser := lss3.NewLeaser() |
| 546 | leaser.SetClient(backend) |
| 547 | leaser.TTL = time.Second |
| 548 | started := time.Now() |
| 549 | lock, e := leaser.AcquireLease(context.Background()) |
| 550 | if e != nil { |
| 551 | t.Fatal(e) |
| 552 | } |
| 553 | rn := startRenewer(leaser, lock, started, 100*time.Millisecond) |
| 554 | defer func() { leaser.ReleaseLease(context.Background(), rn.stop()) }() |
| 555 | initial, e := rn.SafeUntil() |
| 556 | if e != nil { |
| 557 | t.Fatal(e) |
| 558 | } |
| 559 | var began time.Time |
| 560 | select { |
| 561 | case began = <-entered: |
| 562 | case <-time.After(2 * time.Second): |
| 563 | t.Fatal("renewal never started") |
| 564 | } |
| 565 | timeout := time.Now().Add(time.Second) |
| 566 | for { |
| 567 | until, e := rn.SafeUntil() |
| 568 | if e != nil { |
| 569 | t.Fatal(e) |
| 570 | } |
| 571 | if until.After(initial) { |
| 572 | if until.After(began.Add(900 * time.Millisecond)) { |
| 573 | t.Fatal("renewal latency extended deadline") |
| 574 | } |
| 575 | if until == until.Round(0) { |
| 576 | t.Fatal("renewed deadline lost monotonic clock") |
| 577 | } |
| 578 | break |
| 579 | } |
| 580 | if time.Now().After(timeout) { |
| 581 | t.Fatal("renewal never completed") |
| 582 | } |
| 583 | time.Sleep(time.Millisecond) |
| 584 | } |
| 585 | } |
| 586 | |
| 587 | type closeFailureReplica struct { |
| 588 | litestream.ReplicaClient |
| 589 | path string |
| 590 | writes int |
| 591 | failure error |
| 592 | } |
| 593 | |
| 594 | func (c *closeFailureReplica) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) { |
| 595 | c.writes++ |
| 596 | if c.writes > 1 { |
| 597 | return nil, c.failure |
| 598 | } |
| 599 | info, e := c.ReplicaClient.WriteLTXFile(ctx, level, min, max, r) |
| 600 | if e != nil { |
| 601 | return nil, e |
| 602 | } |
| 603 | // Test-only interference after the explicit sync's last upload gives the |
| 604 | // real Close a fresh tail to flush. Maintenance callers must prove stopped |
| 605 | // writers; this deliberately hostile fixture isolates Close error handling. |
| 606 | d, e := sql.Open("sqlite", c.path) |
| 607 | if e != nil { |
| 608 | return nil, e |
| 609 | } |
| 610 | _, e = d.Exec("INSERT INTO t(body) VALUES('late fixture tail')") |
| 611 | closeErr := d.Close() |
| 612 | if e != nil { |
| 613 | return nil, e |
| 614 | } |
| 615 | if closeErr != nil { |
| 616 | return nil, closeErr |
| 617 | } |
| 618 | return info, nil |
| 619 | } |
| 620 | func TestCheckpointChecksCloseAfterSuccessfulSync(t *testing.T) { |
| 621 | backend := &lockStore{} |
| 622 | h := localHeld(t, backend, 10*time.Second) |
| 623 | path := filepath.Join(t.TempDir(), "source.db") |
| 624 | seedDB(t, path) |
| 625 | sentinel := errors.New("synthetic final tail upload failure") |
| 626 | client := &closeFailureReplica{ReplicaClient: lsfile.NewReplicaClient(t.TempDir()), path: path, failure: sentinel} |
| 627 | h.replicaClient = client |
| 628 | ctx, cancel := context.WithTimeout(context.Background(), time.Second) |
| 629 | defer cancel() |
| 630 | receipt, e := h.Checkpoint(ctx, path) |
| 631 | if !errors.Is(e, sentinel) || !receipt.CompletedAt.IsZero() || client.writes < 2 { |
| 632 | t.Fatalf("close error not preserved: %v writes=%d receipt=%+v", e, client.writes, receipt) |
| 633 | } |
| 634 | if e = h.rn.Err(); e != nil { |
| 635 | t.Fatalf("close failure stopped ownership: %v", e) |
| 636 | } |
| 637 | } |
| 638 | |
| 639 | func TestRenewalFailureSignalsLostBeforeDeadline(t *testing.T) { |
| 640 | backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error { |
| 641 | if p.IfMatch != nil { |
| 642 | return &smithy.GenericAPIError{Code: "PreconditionFailed"} |
| 643 | } |
| 644 | return nil |
| 645 | }} |
| 646 | h := localHeld(t, backend, 3*time.Second) |
| 647 | until, e := h.SafeUntil() |
| 648 | if e != nil { |
| 649 | t.Fatal(e) |
| 650 | } |
| 651 | loss := h.Lost() |
| 652 | await(t, loss) |
| 653 | if !time.Now().Before(until) { |
| 654 | t.Fatal("renewal refusal was signaled only at deadline") |
| 655 | } |
| 656 | if e = h.Err(); e == nil { |
| 657 | t.Fatal("loss notification without failed health") |
| 658 | } |
| 659 | if _, e = h.SafeUntil(); e == nil { |
| 660 | t.Fatal("loss notification left safe deadline usable") |
| 661 | } |
| 662 | if h.Lost() != loss { |
| 663 | t.Fatal("loss channel replaced") |
| 664 | } |
| 665 | } |
| 666 | |
| 667 | func TestHealthyReleaseDoesNotSignalLost(t *testing.T) { |
| 668 | h := localHeld(t, &lockStore{}, 300*time.Millisecond) |
| 669 | until, e := h.SafeUntil() |
| 670 | if e != nil { |
| 671 | t.Fatal(e) |
| 672 | } |
| 673 | loss := h.Lost() |
| 674 | if e = h.Release(context.Background()); e != nil { |
| 675 | t.Fatal(e) |
| 676 | } |
| 677 | if _, e = h.SafeUntil(); !errors.Is(e, ErrSpent) { |
| 678 | t.Fatal("released handle remained usable") |
| 679 | } |
| 680 | time.Sleep(time.Until(until) + 30*time.Millisecond) |
| 681 | select { |
| 682 | case <-loss: |
| 683 | t.Fatal("normal release masqueraded as loss") |
| 684 | default: |
| 685 | } |
| 686 | if h.Err() == nil { |
| 687 | t.Fatal("spent handle remained healthy") |
| 688 | } |
| 689 | select { |
| 690 | case <-loss: |
| 691 | t.Fatal("late health check invented loss after release") |
| 692 | default: |
| 693 | } |
| 694 | } |
| 695 | |
amadan