amadan

carlos / platform Public

Expose permanent lease loss notification across handoff

Paul Campbell pushed by paul@keymail.dev 760edce203e20e85b6c8a510f92e3628b428d802 parent f9ba656
3 files changed, +95 −7
  • internal/lease/checkpoint_test.go +71 −1
  • internal/lease/health.go +15 −6
  • internal/lease/lease.go +9 −0
diff --git a/internal/lease/checkpoint_test.go b/internal/lease/checkpoint_test.go
index 2fb063635..b3047e00b 100644
--- a/internal/lease/checkpoint_test.go
+++ b/internal/lease/checkpoint_test.go
@@ -216,6 +216,14 @@ func TestHeldCheckpointVacuumRestoreAndTransfer(t *testing.T) {
t.Fatal(e)
}
defer r.Close(context.Background())
+ if r.Lost() != h.Lost() {
+ t.Fatal("transfer replaced loss channel")
+ }
+ select {
+ case <-r.Lost():
+ t.Fatal("healthy transfer signaled loss")
+ default:
+ }
if r.rn != rn {
t.Fatal("renewer replaced")
}
@@ -237,6 +245,11 @@ func TestHeldCheckpointVacuumRestoreAndTransfer(t *testing.T) {
if _, e = r.SafeUntil(); e == nil {
t.Fatal("stopped deadline accepted")
}
+ select {
+ case <-r.Lost():
+ t.Fatal("normal Close signaled loss")
+ default:
+ }
}
type blockedReplica struct {
@@ -335,7 +348,7 @@ func TestBlockedRenewalDeadlinePrecedesTakeover(t *testing.T) {
}
await(t, entered)
// A worker observes cancellation without waiting for the blocked remote call.
- await(t, h.rn.ctx.Done())
+ await(t, h.Lost())
if e = h.Err(); e == nil {
t.Fatal("blocked renewal remained healthy")
}
@@ -612,3 +625,60 @@ func TestCheckpointChecksCloseAfterSuccessfulSync(t *testing.T) {
t.Fatalf("close failure stopped ownership: %v", e)
}
}
+
+func TestRenewalFailureSignalsLostBeforeDeadline(t *testing.T) {
+ backend := &lockStore{beforePut: func(ctx context.Context, p *s3.PutObjectInput) error {
+ if p.IfMatch != nil {
+ return &smithy.GenericAPIError{Code: "PreconditionFailed"}
+ }
+ return nil
+ }}
+ h := localHeld(t, backend, 3*time.Second)
+ until, e := h.SafeUntil()
+ if e != nil {
+ t.Fatal(e)
+ }
+ loss := h.Lost()
+ await(t, loss)
+ if !time.Now().Before(until) {
+ t.Fatal("renewal refusal was signaled only at deadline")
+ }
+ if e = h.Err(); e == nil {
+ t.Fatal("loss notification without failed health")
+ }
+ if _, e = h.SafeUntil(); e == nil {
+ t.Fatal("loss notification left safe deadline usable")
+ }
+ if h.Lost() != loss {
+ t.Fatal("loss channel replaced")
+ }
+}
+
+func TestHealthyReleaseDoesNotSignalLost(t *testing.T) {
+ h := localHeld(t, &lockStore{}, 300*time.Millisecond)
+ until, e := h.SafeUntil()
+ if e != nil {
+ t.Fatal(e)
+ }
+ loss := h.Lost()
+ if e = h.Release(context.Background()); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = h.SafeUntil(); !errors.Is(e, ErrSpent) {
+ t.Fatal("released handle remained usable")
+ }
+ time.Sleep(time.Until(until) + 30*time.Millisecond)
+ select {
+ case <-loss:
+ t.Fatal("normal release masqueraded as loss")
+ default:
+ }
+ if h.Err() == nil {
+ t.Fatal("spent handle remained healthy")
+ }
+ select {
+ case <-loss:
+ t.Fatal("late health check invented loss after release")
+ default:
+ }
+}
diff --git a/internal/lease/health.go b/internal/lease/health.go
index dca775606..69814f15c 100644
--- a/internal/lease/health.go
+++ b/internal/lease/health.go
@@ -40,29 +40,37 @@ type renewer struct {
done chan struct{}
watchDone chan struct{}
changed chan struct{}
+ lost 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)}
+ 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), lost: make(chan struct{})}
go rn.watch(ctx)
go rn.renew(ctx)
return rn
}
+func (rn *renewer) loseLocked(err error) {
+ if rn.err == nil {
+ rn.err = err
+ close(rn.lost)
+ }
+}
+
// 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
+ if !time.Now().Before(rn.until) {
+ rn.loseLocked(ErrLost)
+ }
+ return rn.err
}
func (rn *renewer) Err() error {
rn.mu.Lock()
@@ -128,7 +136,7 @@ func (rn *renewer) renew(ctx context.Context) {
}
if health == nil {
if err != nil {
- rn.err = err
+ rn.loseLocked(err)
} else {
rn.until = started.Add(rn.leaser.TTL - rn.margin)
rn.checkLocked()
@@ -150,6 +158,7 @@ func (rn *renewer) renew(ctx context.Context) {
func (rn *renewer) stop() *litestream.Lease {
rn.mu.Lock()
+ rn.checkLocked()
rn.stopped = true
rn.mu.Unlock()
rn.cancel()
diff --git a/internal/lease/lease.go b/internal/lease/lease.go
index 84ab47bdd..8755c2ed3 100644
--- a/internal/lease/lease.go
+++ b/internal/lease/lease.go
@@ -178,6 +178,11 @@ func (h *Held) Err() error {
return h.rn.Err()
}
+// Lost closes once on renewal failure or the safety deadline. Healthy transfer
+// and explicit release do not close it. A worker must recheck Err/SafeUntil and
+// prove its process tree stopped before the deadline; notification is not proof.
+func (h *Held) Lost() <-chan struct{} { return h.rn.lost }
+
// 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) {
@@ -340,6 +345,10 @@ func (r *Replicator) Err() error { return r.rn.Err() }
func (r *Replicator) SafeUntil() (time.Time, error) { return r.rn.SafeUntil() }
+// Lost is the same loss notification channel held before replication began.
+// Normal Close does not signal loss; SafeUntil refuses the stopped handle.
+func (r *Replicator) Lost() <-chan struct{} { return r.rn.lost }
+
// 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