amadan

carlos / platform Public

Retain lease ownership through exclusive checkpoints

Paul Campbell pushed by paul@keymail.dev f9ba656bc36c3232f3f5a57392f41d0ecab74ce2 parent 6ea3646
4 files changed, +951 −104
  • internal/lease/checkpoint.go +56 −0
  • internal/lease/checkpoint_test.go +614 −0
  • internal/lease/health.go +161 −0
  • internal/lease/lease.go +120 −104
diff --git a/internal/lease/checkpoint.go b/internal/lease/checkpoint.go
new file mode 100644
index 000000000..a7d7d1ec9
--- /dev/null
+++ b/internal/lease/checkpoint.go
@@ -0,0 +1,56 @@
+package lease
+
+import (
+ "context"
+ "errors"
+ "time"
+)
+
+// CheckpointReceipt describes a completed sync and close under a healthy
+// lease. It does not prove restored database semantics or authorize a command.
+// SafeUntil is useful only in-process, with its monotonic clock component.
+type CheckpointReceipt struct {
+ CompletedAt time.Time
+ SafeUntil time.Time
+}
+
+// Checkpoint keeps the same renewal running while one temporary replica store
+// opens, syncs and closes. Failure poisons the handle but does not release it;
+// the caller must retain its fence, record the failure and explicitly Release.
+func (h *Held) Checkpoint(ctx context.Context, dbPath string) (receipt CheckpointReceipt, err error) {
+ if err = h.begin(false); err != nil {
+ return receipt, err
+ }
+ defer func() { h.finish(false, err) }()
+ opCtx, cancel := context.WithCancel(ctx)
+ stop := context.AfterFunc(h.rn.ctx, cancel)
+ defer stop()
+ defer cancel()
+ store, db := h.newStore(dbPath, 0)
+ // Short checkpoints need only the explicit L0 sync. Background snapshots
+ // are unnecessary and v0.5.14 closes the DB before joining those monitors.
+ db.MonitorInterval = 0
+ db.Replica.MonitorEnabled = false
+ store.CompactionMonitorEnabled = false
+ store.L0RetentionCheckInterval = 0
+ store.SetRetentionEnabled(false)
+ openErr := store.Open(opCtx)
+ var syncErr error
+ if openErr == nil {
+ syncErr = db.SyncAndWait(opCtx)
+ }
+ if openErr != nil || syncErr != nil {
+ cancel()
+ }
+ closeErr := store.Close(opCtx)
+ until, healthErr := h.rn.SafeUntil()
+ err = errors.Join(openErr, syncErr, closeErr, healthErr, opCtx.Err())
+ if err != nil {
+ return CheckpointReceipt{}, err
+ }
+ completed := time.Now()
+ if !completed.Before(until) {
+ return CheckpointReceipt{}, ErrLost
+ }
+ return CheckpointReceipt{CompletedAt: completed, SafeUntil: until}, nil
+}
diff --git a/internal/lease/checkpoint_test.go b/internal/lease/checkpoint_test.go
new file mode 100644
index 000000000..2fb063635
--- /dev/null
+++ b/internal/lease/checkpoint_test.go
@@ -0,0 +1,614 @@
+package lease
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ s3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/aws/smithy-go"
+ "github.com/benbjohnson/litestream"
+ lsfile "github.com/benbjohnson/litestream/file"
+ lss3 "github.com/benbjohnson/litestream/s3"
+ "github.com/superfly/ltx"
+)
+
+type lockStore struct {
+ mu sync.Mutex
+ body []byte
+ etag string
+ puts, deletes int
+ beforePut func(context.Context, *s3.PutObjectInput) error
+}
+
+func (s *lockStore) GetObject(ctx context.Context, p *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if s.body == nil {
+ return nil, &smithy.GenericAPIError{Code: "NoSuchKey"}
+ }
+ return &s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader(bytes.Clone(s.body))), ETag: aws.String(s.etag)}, nil
+}
+func (s *lockStore) PutObject(ctx context.Context, p *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
+ raw, e := io.ReadAll(p.Body)
+ if e != nil {
+ return nil, e
+ }
+ if s.beforePut != nil {
+ if e := s.beforePut(ctx, p); e != nil {
+ return nil, e
+ }
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if (aws.ToString(p.IfNoneMatch) == "*" && s.body != nil) || (p.IfMatch != nil && aws.ToString(p.IfMatch) != s.etag) {
+ return nil, &smithy.GenericAPIError{Code: "PreconditionFailed"}
+ }
+ s.puts++
+ s.etag = fmt.Sprintf("\"etag-%d\"", s.puts)
+ s.body = raw
+ return &s3.PutObjectOutput{ETag: aws.String(s.etag)}, nil
+}
+func (s *lockStore) DeleteObject(ctx context.Context, p *s3.DeleteObjectInput, _ ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if p.IfMatch == nil || aws.ToString(p.IfMatch) != s.etag {
+ return nil, &smithy.GenericAPIError{Code: "PreconditionFailed"}
+ }
+ s.deletes++
+ s.body = nil
+ s.etag = ""
+ return &s3.DeleteObjectOutput{}, nil
+}
+func (s *lockStore) counts() (int, int) { s.mu.Lock(); defer s.mu.Unlock(); return s.puts, s.deletes }
+func (s *lockStore) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ var err error
+ switch r.Method {
+ case "GET":
+ var out *s3.GetObjectOutput
+ out, err = s.GetObject(r.Context(), &s3.GetObjectInput{})
+ if err == nil {
+ w.Header().Set("ETag", aws.ToString(out.ETag))
+ io.Copy(w, out.Body)
+ return
+ }
+ case "PUT":
+ p := &s3.PutObjectInput{Body: r.Body}
+ if v := r.Header.Get("If-Match"); v != "" {
+ p.IfMatch = aws.String(v)
+ }
+ if v := r.Header.Get("If-None-Match"); v != "" {
+ p.IfNoneMatch = aws.String(v)
+ }
+ var out *s3.PutObjectOutput
+ out, err = s.PutObject(r.Context(), p)
+ if err == nil {
+ w.Header().Set("ETag", aws.ToString(out.ETag))
+ return
+ }
+ case "DELETE":
+ _, err = s.DeleteObject(r.Context(), &s3.DeleteObjectInput{IfMatch: aws.String(r.Header.Get("If-Match"))})
+ if err == nil {
+ w.WriteHeader(204)
+ return
+ }
+ }
+ var api *smithy.GenericAPIError
+ if errors.As(err, &api) {
+ code := 412
+ if api.Code == "NoSuchKey" {
+ code = 404
+ }
+ w.Header().Set("Content-Type", "application/xml")
+ w.WriteHeader(code)
+ fmt.Fprintf(w, "<Error><Code>%s</Code></Error>", api.Code)
+ return
+ }
+ w.WriteHeader(500)
+}
+func localHeld(t *testing.T, backend *lockStore, ttl time.Duration) *Held {
+ t.Helper()
+ srv := httptest.NewServer(backend)
+ t.Cleanup(srv.Close)
+ fakeCreds(t)
+ t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
+ h, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl)
+ if e != nil {
+ t.Fatal(e)
+ }
+ t.Cleanup(func() { _ = h.Release(context.Background()) })
+ return h
+}
+func await(t *testing.T, ch <-chan struct{}) {
+ t.Helper()
+ select {
+ case <-ch:
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out")
+ }
+}
+func restoreCount(t *testing.T, replica, out string) int {
+ t.Helper()
+ r := litestream.NewReplicaWithClient(nil, lsfile.NewReplicaClient(replica))
+ opt := litestream.NewRestoreOptions()
+ opt.OutputPath = out
+ if e := r.Restore(context.Background(), opt); e != nil {
+ t.Fatal(e)
+ }
+ d, e := sql.Open("sqlite", "file:"+out+"?mode=ro")
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer d.Close()
+ var n int
+ var integrity string
+ if e = d.QueryRow("PRAGMA integrity_check").Scan(&integrity); e != nil || integrity != "ok" {
+ t.Fatalf("integrity %q %v", integrity, e)
+ }
+ if e = d.QueryRow("SELECT count(*) FROM t").Scan(&n); e != nil {
+ t.Fatal(e)
+ }
+ var contents string
+ if e = d.QueryRow("SELECT group_concat(body, '|') FROM (SELECT body FROM t ORDER BY id)").Scan(&contents); e != nil {
+ t.Fatal(e)
+ }
+ if n == 5 && contents != "post-vacuum|post-vacuum|post-vacuum|post-vacuum|post-vacuum" {
+ t.Fatalf("restored content receipt differs: %q", contents)
+ }
+ return n
+}
+func TestHeldCheckpointVacuumRestoreAndTransfer(t *testing.T) {
+ backend := &lockStore{}
+ h := localHeld(t, backend, 10*time.Second)
+ path := filepath.Join(t.TempDir(), "source.db")
+ seedDB(t, path)
+ replica := t.TempDir()
+ h.replicaClient = lsfile.NewReplicaClient(replica)
+ rn := h.rn
+ receipt, e := h.Checkpoint(context.Background(), path)
+ if e != nil {
+ t.Fatal(e)
+ }
+ if receipt.CompletedAt.IsZero() || !receipt.CompletedAt.Before(receipt.SafeUntil) || h.rn != rn {
+ t.Fatal("invalid checkpoint receipt/ownership")
+ }
+ if n := restoreCount(t, replica, filepath.Join(t.TempDir(), "old.db")); n != 50 {
+ t.Fatal(n)
+ }
+ d, e := sql.Open("sqlite", path)
+ if e != nil {
+ t.Fatal(e)
+ }
+ _, e = d.Exec("DELETE FROM t WHERE id>5; UPDATE t SET body='post-vacuum'; PRAGMA wal_checkpoint(TRUNCATE); VACUUM; PRAGMA wal_checkpoint(TRUNCATE)")
+ if e != nil {
+ t.Fatal(e)
+ }
+ if e = d.Close(); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = h.Checkpoint(context.Background(), path); e != nil {
+ t.Fatal(e)
+ }
+ if n := restoreCount(t, replica, filepath.Join(t.TempDir(), "new.db")); n != 5 {
+ t.Fatalf("stale replica: %d", n)
+ }
+ if _, deletes := backend.counts(); deletes != 0 {
+ t.Fatal("checkpoint released lock")
+ }
+ // The actual leaser sees the same still-owned lock after both checkpoints.
+ if peer, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 10*time.Second); e == nil {
+ peer.Release(context.Background())
+ t.Fatal("acquisition gap")
+ }
+ r, e := h.Replicate(context.Background(), path, 0)
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer r.Close(context.Background())
+ if r.rn != rn {
+ t.Fatal("renewer replaced")
+ }
+ if e = h.Release(context.Background()); !errors.Is(e, ErrSpent) {
+ t.Fatalf("spent release %v", e)
+ }
+ if _, e = h.SafeUntil(); !errors.Is(e, ErrSpent) {
+ t.Fatal("spent deadline accepted")
+ }
+ if _, e = r.SafeUntil(); e != nil {
+ t.Fatal(e)
+ }
+ if _, deletes := backend.counts(); deletes != 0 {
+ t.Fatal("spent Held released successor")
+ }
+ if e = r.Close(context.Background()); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = r.SafeUntil(); e == nil {
+ t.Fatal("stopped deadline accepted")
+ }
+}
+
+type blockedReplica struct {
+ litestream.ReplicaClient
+ entered chan struct{}
+ release chan struct{}
+ once sync.Once
+ fail error
+}
+
+func (b *blockedReplica) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
+ b.once.Do(func() { close(b.entered) })
+ select {
+ case <-b.release:
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+ if b.fail != nil {
+ return nil, b.fail
+ }
+ return b.ReplicaClient.WriteLTXFile(ctx, level, min, max, r)
+}
+func TestCheckpointExclusionAndPoison(t *testing.T) {
+ for _, fail := range []bool{false, true} {
+ t.Run(fmt.Sprint(fail), func(t *testing.T) {
+ backend := &lockStore{}
+ h := localHeld(t, backend, 10*time.Second)
+ path := filepath.Join(t.TempDir(), "source.db")
+ seedDB(t, path)
+ b := &blockedReplica{ReplicaClient: lsfile.NewReplicaClient(t.TempDir()), entered: make(chan struct{}), release: make(chan struct{})}
+ if fail {
+ b.fail = errors.New("synthetic replica write refusal")
+ }
+ h.replicaClient = b
+ opCtx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { _, e := h.Checkpoint(opCtx, path); done <- e }()
+ await(t, b.entered)
+ if _, e := h.Checkpoint(context.Background(), path); !errors.Is(e, ErrBusy) {
+ t.Fatal(e)
+ }
+ if _, e := h.Replicate(context.Background(), path, 0); !errors.Is(e, ErrBusy) {
+ t.Fatal(e)
+ }
+ if e := h.Release(context.Background()); !errors.Is(e, ErrBusy) {
+ t.Fatal(e)
+ }
+ close(b.release)
+ e := <-done
+ if fail {
+ if e == nil {
+ t.Fatal("failed sync accepted")
+ }
+ if _, e = h.Checkpoint(context.Background(), path); e == nil {
+ t.Fatal("poisoned checkpoint")
+ }
+ if _, e = h.Replicate(context.Background(), path, 0); e == nil {
+ t.Fatal("poisoned replicate")
+ }
+ if _, e = h.SafeUntil(); e == nil {
+ t.Fatal("poisoned command deadline")
+ }
+ } else if e != nil {
+ t.Fatal(e)
+ }
+ if e = h.rn.Err(); e != nil {
+ t.Fatalf("ownership released on checkpoint outcome: %v", e)
+ }
+ if _, deletes := backend.counts(); deletes != 0 {
+ t.Fatal("unexpected release")
+ }
+ if e = h.Release(context.Background()); e != nil {
+ t.Fatal(e)
+ }
+ })
+ }
+}
+
+func TestBlockedRenewalDeadlinePrecedesTakeover(t *testing.T) {
+ entered := make(chan struct{})
+ var once sync.Once
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch != nil {
+ once.Do(func() { close(entered) })
+ <-ctx.Done()
+ return ctx.Err()
+ }
+ return nil
+ }}
+ ttl := 900 * time.Millisecond
+ h := localHeld(t, backend, ttl)
+ until, e := h.SafeUntil()
+ if e != nil {
+ t.Fatal(e)
+ }
+ await(t, entered)
+ // A worker observes cancellation without waiting for the blocked remote call.
+ await(t, h.rn.ctx.Done())
+ if e = h.Err(); e == nil {
+ t.Fatal("blocked renewal remained healthy")
+ }
+ if time.Since(until) > ttl/10 {
+ t.Fatal("loss notification consumed termination margin")
+ }
+ backend.mu.Lock()
+ var original litestream.Lease
+ json.Unmarshal(backend.body, &original)
+ backend.mu.Unlock()
+ if !time.Now().Before(original.ExpiresAt) {
+ t.Fatal("termination did not precede lease expiry")
+ }
+ if peer, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl); e == nil {
+ peer.Release(context.Background())
+ t.Fatal("peer took live lock")
+ }
+ time.Sleep(time.Until(original.ExpiresAt) + 20*time.Millisecond)
+ // A separate real leaser can take over the expired CAS object. Its acquisition
+ // uses If-Match, so disable the synthetic blocked-renewal hook first.
+ backend.beforePut = nil
+ peer, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl)
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer peer.Release(context.Background())
+ if e = h.Release(context.Background()); e == nil {
+ t.Fatal("stale release erased peer")
+ }
+ if e = peer.Err(); e != nil {
+ t.Fatal(e)
+ }
+}
+
+func TestLateRenewalCannotResurrect(t *testing.T) {
+ gate := make(chan struct{})
+ entered := make(chan struct{})
+ var once sync.Once
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch != nil {
+ once.Do(func() { close(entered) })
+ <-gate
+ }
+ return nil
+ }}
+ leaser := lss3.NewLeaser()
+ leaser.SetClient(backend)
+ leaser.TTL = 300 * time.Millisecond
+ started := time.Now()
+ lock, e := leaser.AcquireLease(context.Background())
+ if e != nil {
+ t.Fatal(e)
+ }
+ rn := startRenewer(leaser, lock, started, 30*time.Millisecond)
+ await(t, entered)
+ await(t, rn.ctx.Done())
+ if !errors.Is(rn.Err(), ErrLost) {
+ t.Fatal("deadline not latched")
+ }
+ close(gate)
+ await(t, rn.done)
+ if !errors.Is(rn.Err(), ErrLost) {
+ t.Fatal("late acknowledgement resurrected ownership")
+ }
+ if _, e = rn.SafeUntil(); e == nil {
+ t.Fatal("late acknowledgement extended deadline")
+ }
+ leaser.ReleaseLease(context.Background(), rn.stop())
+}
+
+func TestAcquireDeadlineStartsBeforeRemoteRequest(t *testing.T) {
+ entered := make(chan time.Time, 1)
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch == nil {
+ entered <- time.Now()
+ time.Sleep(120 * time.Millisecond)
+ }
+ return nil
+ }}
+ ttl := time.Second
+ h := localHeld(t, backend, ttl)
+ beforeAck := <-entered
+ until, e := h.SafeUntil()
+ if e != nil {
+ t.Fatal(e)
+ }
+ if until.After(beforeAck.Add(ttl - marginForTTL(ttl))) {
+ t.Fatal("acquisition latency extended safe deadline")
+ }
+ if until == until.Round(0) {
+ t.Fatal("deadline lost monotonic clock")
+ }
+}
+
+func TestAcquireLateSuccessGrantsNoHandle(t *testing.T) {
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch == nil {
+ time.Sleep(350 * time.Millisecond)
+ }
+ return nil
+ }}
+ srv := httptest.NewServer(backend)
+ defer srv.Close()
+ fakeCreds(t)
+ t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
+ h, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 300*time.Millisecond)
+ if e == nil || h != nil {
+ if h != nil {
+ h.Release(context.Background())
+ }
+ t.Fatal("late acquisition granted authority")
+ }
+ if _, deletes := backend.counts(); deletes != 1 {
+ t.Fatal("known late acquisition was not conditionally cleaned up")
+ }
+}
+
+func TestCheckpointLossCancelsBlockedSync(t *testing.T) {
+ entered := make(chan struct{})
+ var once sync.Once
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch != nil {
+ once.Do(func() { close(entered) })
+ <-ctx.Done()
+ return ctx.Err()
+ }
+ return nil
+ }}
+ h := localHeld(t, backend, 600*time.Millisecond)
+ path := filepath.Join(t.TempDir(), "source.db")
+ seedDB(t, path)
+ b := &blockedReplica{ReplicaClient: lsfile.NewReplicaClient(t.TempDir()), entered: make(chan struct{}), release: make(chan struct{})}
+ h.replicaClient = b
+ done := make(chan error, 1)
+ go func() { _, e := h.Checkpoint(context.Background(), path); done <- e }()
+ await(t, b.entered)
+ await(t, entered)
+ select {
+ case e := <-done:
+ if e == nil {
+ t.Fatal("lost lease returned receipt")
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("loss did not cancel checkpoint")
+ }
+ if _, deletes := backend.counts(); deletes != 0 {
+ t.Fatal("checkpoint released after loss")
+ }
+ if _, e := h.SafeUntil(); e == nil {
+ t.Fatal("failed checkpoint usable")
+ }
+}
+
+func TestCheckpointOpenFailureLeavesOwnedPoisonedHandle(t *testing.T) {
+ backend := &lockStore{}
+ h := localHeld(t, backend, 10*time.Second)
+ // A directory is never a SQLite database; the real store Open must refuse.
+ if receipt, e := h.Checkpoint(context.Background(), t.TempDir()); e == nil || !receipt.CompletedAt.IsZero() {
+ t.Fatal("failed open returned receipt")
+ }
+ if e := h.rn.Err(); e != nil {
+ t.Fatal("open failure lost held renewal")
+ }
+ if _, e := h.Replicate(context.Background(), "unused", 0); e == nil {
+ t.Fatal("open failure did not poison")
+ }
+ if _, deletes := backend.counts(); deletes != 0 {
+ t.Fatal("open failure released lease")
+ }
+}
+
+func TestRenewDeadlineStartsBeforeRemoteRequest(t *testing.T) {
+ entered := make(chan time.Time, 1)
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch != nil {
+ select {
+ case entered <- time.Now():
+ default:
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ return nil
+ }}
+ leaser := lss3.NewLeaser()
+ leaser.SetClient(backend)
+ leaser.TTL = time.Second
+ started := time.Now()
+ lock, e := leaser.AcquireLease(context.Background())
+ if e != nil {
+ t.Fatal(e)
+ }
+ rn := startRenewer(leaser, lock, started, 100*time.Millisecond)
+ defer func() { leaser.ReleaseLease(context.Background(), rn.stop()) }()
+ initial, e := rn.SafeUntil()
+ if e != nil {
+ t.Fatal(e)
+ }
+ var began time.Time
+ select {
+ case began = <-entered:
+ case <-time.After(2 * time.Second):
+ t.Fatal("renewal never started")
+ }
+ timeout := time.Now().Add(time.Second)
+ for {
+ until, e := rn.SafeUntil()
+ if e != nil {
+ t.Fatal(e)
+ }
+ if until.After(initial) {
+ if until.After(began.Add(900 * time.Millisecond)) {
+ t.Fatal("renewal latency extended deadline")
+ }
+ if until == until.Round(0) {
+ t.Fatal("renewed deadline lost monotonic clock")
+ }
+ break
+ }
+ if time.Now().After(timeout) {
+ t.Fatal("renewal never completed")
+ }
+ time.Sleep(time.Millisecond)
+ }
+}
+
+type closeFailureReplica struct {
+ litestream.ReplicaClient
+ path string
+ writes int
+ failure error
+}
+
+func (c *closeFailureReplica) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
+ c.writes++
+ if c.writes > 1 {
+ return nil, c.failure
+ }
+ info, e := c.ReplicaClient.WriteLTXFile(ctx, level, min, max, r)
+ if e != nil {
+ return nil, e
+ }
+ // Test-only interference after the explicit sync's last upload gives the
+ // real Close a fresh tail to flush. Maintenance callers must prove stopped
+ // writers; this deliberately hostile fixture isolates Close error handling.
+ d, e := sql.Open("sqlite", c.path)
+ if e != nil {
+ return nil, e
+ }
+ _, e = d.Exec("INSERT INTO t(body) VALUES('late fixture tail')")
+ closeErr := d.Close()
+ if e != nil {
+ return nil, e
+ }
+ if closeErr != nil {
+ return nil, closeErr
+ }
+ return info, nil
+}
+func TestCheckpointChecksCloseAfterSuccessfulSync(t *testing.T) {
+ backend := &lockStore{}
+ h := localHeld(t, backend, 10*time.Second)
+ path := filepath.Join(t.TempDir(), "source.db")
+ seedDB(t, path)
+ sentinel := errors.New("synthetic final tail upload failure")
+ client := &closeFailureReplica{ReplicaClient: lsfile.NewReplicaClient(t.TempDir()), path: path, failure: sentinel}
+ h.replicaClient = client
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ receipt, e := h.Checkpoint(ctx, path)
+ if !errors.Is(e, sentinel) || !receipt.CompletedAt.IsZero() || client.writes < 2 {
+ t.Fatalf("close error not preserved: %v writes=%d receipt=%+v", e, client.writes, receipt)
+ }
+ if e = h.rn.Err(); e != nil {
+ t.Fatalf("close failure stopped ownership: %v", e)
+ }
+}
diff --git a/internal/lease/health.go b/internal/lease/health.go
new file mode 100644
index 000000000..dca775606
--- /dev/null
+++ b/internal/lease/health.go
@@ -0,0 +1,161 @@
+package lease
+
+import (
+ "context"
+ "errors"
+ "sync"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ lss3 "github.com/benbjohnson/litestream/s3"
+)
+
+var (
+ ErrLost = errors.New("lease safety deadline elapsed")
+ ErrSpent = errors.New("lease handle is spent")
+ ErrBusy = errors.New("lease operation already running")
+)
+
+const terminationMargin = 5 * time.Second
+
+// Production uses the fixed margin. Short TTLs exist only through acquireTTL,
+// the package's private test seam.
+func marginForTTL(ttl time.Duration) time.Duration {
+ if ttl < 3*terminationMargin {
+ return ttl / 10
+ }
+ return terminationMargin
+}
+
+type renewer struct {
+ ctx context.Context
+ leaser *lss3.Leaser
+ mu sync.Mutex
+ lease *litestream.Lease
+ err error
+ until time.Time
+ margin time.Duration
+ stopped bool
+ cancel context.CancelFunc
+ done chan struct{}
+ watchDone chan struct{}
+ changed chan struct{}
+}
+
+func startRenewer(leaser *lss3.Leaser, lease *litestream.Lease, started time.Time, margin time.Duration) *renewer {
+ ctx, cancel := context.WithCancel(context.Background())
+ rn := &renewer{ctx: ctx, leaser: leaser, lease: lease, until: started.Add(leaser.TTL - margin), margin: margin, cancel: cancel, done: make(chan struct{}), watchDone: make(chan struct{}), changed: make(chan struct{}, 1)}
+ go rn.watch(ctx)
+ go rn.renew(ctx)
+ return rn
+}
+
+// checkLocked latches loss permanently; an acknowledgement arriving after a
+// blocked renewal cannot resurrect ownership, even before the timer runs.
+func (rn *renewer) checkLocked() error {
+ if rn.err == nil && !time.Now().Before(rn.until) {
+ rn.err = ErrLost
+ }
+ if rn.err != nil {
+ return rn.err
+ }
+ if rn.stopped {
+ return ErrSpent
+ }
+ return nil
+}
+func (rn *renewer) Err() error {
+ rn.mu.Lock()
+ defer rn.mu.Unlock()
+ return rn.checkLocked()
+}
+func (rn *renewer) SafeUntil() (time.Time, error) {
+ rn.mu.Lock()
+ defer rn.mu.Unlock()
+ if err := rn.checkLocked(); err != nil {
+ return time.Time{}, err
+ }
+ return rn.until, nil
+}
+func (rn *renewer) watch(ctx context.Context) {
+ defer close(rn.watchDone)
+ for {
+ until, err := rn.SafeUntil()
+ if err != nil {
+ rn.cancel()
+ return
+ }
+ timer := time.NewTimer(time.Until(until))
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ return
+ case <-rn.changed:
+ timer.Stop()
+ case <-timer.C:
+ }
+ }
+}
+func (rn *renewer) renew(ctx context.Context) {
+ defer close(rn.done)
+ every := rn.leaser.TTL / 3
+ if every <= 0 {
+ every = time.Nanosecond
+ }
+ ticker := time.NewTicker(every)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ rn.mu.Lock()
+ err := rn.checkLocked()
+ cur := rn.lease
+ rn.mu.Unlock()
+ if err != nil {
+ rn.cancel()
+ return
+ }
+ started := time.Now()
+ renewed, err := rn.leaser.RenewLease(ctx, cur)
+ rn.mu.Lock()
+ health := rn.checkLocked()
+ // Retain the newest ETag solely for conditional cleanup, including a late
+ // success. It never repairs a latched loss or extends a stopped handle.
+ if err == nil {
+ rn.lease = renewed
+ }
+ if health == nil {
+ if err != nil {
+ rn.err = err
+ } else {
+ rn.until = started.Add(rn.leaser.TTL - rn.margin)
+ rn.checkLocked()
+ }
+ }
+ failed := rn.err != nil || rn.stopped
+ rn.mu.Unlock()
+ if failed {
+ rn.cancel()
+ return
+ }
+ select {
+ case rn.changed <- struct{}{}:
+ default:
+ }
+ }
+ }
+}
+
+func (rn *renewer) stop() *litestream.Lease {
+ rn.mu.Lock()
+ rn.stopped = true
+ rn.mu.Unlock()
+ rn.cancel()
+ <-rn.done
+ <-rn.watchDone
+ rn.mu.Lock()
+ defer rn.mu.Unlock()
+ return rn.lease
+}
diff --git a/internal/lease/lease.go b/internal/lease/lease.go
index 3a108ed81..84ab47bdd 100644
--- a/internal/lease/lease.go
+++ b/internal/lease/lease.go
@@ -92,80 +92,6 @@ func (e *HeldError) Error() string {
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{})}
- // Renew at a third of the lease's OWN TTL, not the package constant: the
- // leaser is the one that stamps expiry (leaser.TTL), so reading the
- // interval off it keeps renewal and expiry in step for any TTL. In
- // production leaser.TTL is always LeaseTTL, so this is LeaseTTL/3 as
- // before; under a test's short TTL it is what makes "stopped renewing"
- // distinguishable from "never renewed at all". The fallback only guards
- // NewTicker's panic on a non-positive duration.
- renewEvery := leaser.TTL / 3
- if renewEvery <= 0 {
- renewEvery = LeaseTTL / 3
- }
- go func() {
- defer close(rn.done)
- t := time.NewTicker(renewEvery)
- 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
@@ -173,9 +99,14 @@ func (rn *renewer) stop() *litestream.Lease {
// 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
+ leaser *lss3.Leaser
+ region string // the leaser's client keeps its region private; carry it for Replicate
+ rn *renewer
+ mu sync.Mutex
+ busy bool
+ spent bool
+ poison error
+ replicaClient litestream.ReplicaClient // private local-replica test seam
}
// Acquire claims the single-writer lease at s3://bucket/prefix/lock.json
@@ -212,6 +143,7 @@ func acquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Dur
leaser.Path = prefix // lock lives at <prefix>/lock.json, colocated with the replica
leaser.TTL = ttl
+ started := time.Now()
lease, err := leaser.AcquireLease(ctx)
if err != nil {
// A *litestream.LeaseExistsError means someone else holds it —
@@ -222,15 +154,81 @@ func acquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Dur
}
return nil, fmt.Errorf("acquire lease: %w", err)
}
- return &Held{leaser: leaser, region: region, rn: startRenewer(leaser, lease)}, nil
+ rn := startRenewer(leaser, lease, started, marginForTTL(ttl))
+ if err := rn.Err(); err != nil {
+ cur := rn.stop()
+ relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ _ = leaser.ReleaseLease(relCtx, cur)
+ return nil, err
+ }
+ return &Held{leaser: leaser, region: region, rn: rn}, nil
}
-// Err reports whether lease renewal has failed (leadership lost).
-func (h *Held) Err() error { return h.rn.Err() }
+// Err refuses a spent or failed handle as well as lost ownership.
+func (h *Held) Err() error {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.spent {
+ return ErrSpent
+ }
+ if h.poison != nil {
+ return h.poison
+ }
+ return h.rn.Err()
+}
+
+// SafeUntil retains Go's monotonic clock component. It must not be serialized
+// and later treated as ownership evidence; callers must keep checking health.
+func (h *Held) SafeUntil() (time.Time, error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.spent {
+ return time.Time{}, ErrSpent
+ }
+ if h.poison != nil {
+ return time.Time{}, h.poison
+ }
+ return h.rn.SafeUntil()
+}
+
+func (h *Held) begin(release bool) error {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.spent {
+ return ErrSpent
+ }
+ if h.busy {
+ return ErrBusy
+ }
+ if !release {
+ if h.poison != nil {
+ return h.poison
+ }
+ if err := h.rn.Err(); err != nil {
+ return err
+ }
+ }
+ h.busy = true
+ return nil
+}
+func (h *Held) finish(spent bool, poison error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.busy = false
+ h.spent = spent
+ if poison != nil {
+ h.poison = poison
+ }
+}
-// Release stops renewal and releases the lease — the bail-out when a wake
-// fails between acquisition and replication.
+// Release is the only operation allowed after a failed checkpoint. A failed
+// remote release still spends this handle; it cannot authorize more work.
func (h *Held) Release(ctx context.Context) error {
+ if err := h.begin(true); err != nil {
+ return err
+ }
+ defer h.finish(true, nil)
lease := h.rn.stop()
if lease == nil {
return nil
@@ -252,36 +250,47 @@ func (h *Held) Release(ctx context.Context) error {
// 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.begin(false); err != nil {
+ return nil, err
+ }
+ transferred := false
+ defer func() { h.finish(transferred, nil) }()
+ opCtx, cancel := context.WithCancel(ctx)
+ stop := context.AfterFunc(h.rn.ctx, cancel)
+ defer stop()
+ defer cancel()
+ store, db := h.newStore(dbPath, syncInterval)
+ if err := store.Open(opCtx); err != nil {
+ cancel()
+ closeErr := store.Close(opCtx)
+ return nil, errors.Join(fmt.Errorf("open litestream store: %w", err), closeErr)
+ }
if err := h.rn.Err(); err != nil {
- return nil, fmt.Errorf("lease already lost: %w", err)
+ cancel()
+ closeErr := store.Close(opCtx)
+ return nil, errors.Join(err, closeErr)
}
+ transferred = true
+ return &Replicator{leaser: h.leaser, store: store, db: db, rn: h.rn}, nil
+}
+func (h *Held) newStore(dbPath string, syncInterval time.Duration) (*litestream.Store, *litestream.DB) {
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.
+ // Preserve the activator namespace: sharing the default metadata directory
+ // with another replicator corrupts shadow state even across sequential runs.
db.SetMetaPath(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
db.MonitorInterval = dbMonitorInterval
- client := newReplicaClient(h.leaser.Bucket, h.leaser.Path, h.region)
+ client := h.replicaClient
+ if client == nil {
+ client = newReplicaClient(h.leaser.Bucket, h.leaser.Path, 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
+ return store, db
}
// Replicator couples an exclusive S3 lease with a running litestream
@@ -289,10 +298,15 @@ func (h *Held) Replicate(ctx context.Context, dbPath string, syncInterval time.D
// 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
+ leaser *lss3.Leaser
+ store *litestream.Store
+ db *litestream.DB
+ rn *renewer
+ mu sync.Mutex
+ busy bool
+ spent bool
+ poison error
+ replicaClient litestream.ReplicaClient // private local-replica test seam
}
// ReplicateWithLease acquires the single-writer lease at
@@ -324,6 +338,8 @@ func (r *Replicator) SyncAndWait(ctx context.Context) error {
// Err reports whether lease renewal has failed (leadership lost).
func (r *Replicator) Err() error { return r.rn.Err() }
+func (r *Replicator) SafeUntil() (time.Time, error) { return r.rn.SafeUntil() }
+
// 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