diff --git a/cmd/mesh-controller/acting.go b/cmd/mesh-controller/acting.go index fe5aca76..ebf60696 100644 --- a/cmd/mesh-controller/acting.go +++ b/cmd/mesh-controller/acting.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/lease" "github.com/novox/mesh-controller/internal/link" @@ -72,6 +73,14 @@ type actor struct { // lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given // the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it. renewEvery, poll time.Duration + // releasing is held for the whole of a release, so a second caller — the stop signal's and serve's own + // deferred one — waits until the lease is given back and its epoch recorded, never past it to the store's + // close (novox/hq issue 906). + releasing sync.Mutex + // keeper and inv are where a takeover is said (sayTakeover): set once the serving controller's condition + // store opens; nil before, and in a command. + keeper *conditions.Keeper + inv *inventory.Inventory } // theLease is this process's standing to the lease. @@ -174,8 +183,11 @@ func (a *actor) forACommand(ctx context.Context) (uint64, error) { return holder.Epoch, nil } -// release gives back what this process holds, at its end. +// release gives back what this process holds, at its end — or at the stop signal, before the rest of the +// process is taken down (serveUnderTheLease). A second call waits for the first to finish. func (a *actor) release() { + a.releasing.Lock() + defer a.releasing.Unlock() a.mu.Lock() closing := a.close a.close = nil @@ -266,20 +278,40 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory } a.mu.Lock() a.close = func() { - if held, err := l.Epoch(); err == nil { + // Given back first, then recorded: an epoch is recorded released only when its key is gone, and + // whatever the gate says by then — a shutdown longer than half the key's age closed it, and the epoch + // was left open for the next holder to find "expired" (novox/hq issue 906, 2026-10-26). + if given := l.Release(context.Background()); given != 0 { ending, cancel := context.WithTimeout(context.Background(), 5*time.Second) - if err := inv.EndEpoch(ending, held, inventory.EpochReleased); err != nil { - say("how epoch %d ended could not be recorded: %v", held, err) + if err := inv.EndEpoch(ending, given, inventory.EpochReleased); err != nil { + say("how epoch %d ended could not be recorded: %v", given, err) } cancel() } - l.Release(context.Background()) js.Close() } a.mu.Unlock() + // **The stop signal gives the lease back first** (novox/hq issue 906): the moment ctx ends, before the rest + // of the process is taken down, the gate closes and the key goes, so the next controller takes the lease + // at once and the epoch is recorded released. Waiting for the process's own end left the key unrenewed for + // as long as the shutdown took — ten seconds on 2026-10-26, past the gate's half-age. + go func() { + <-ctx.Done() + if a.holds(l) { + say("stopping: the controller lease is given back first") + } + a.release() + }() return gone, nil } +// holds is whether this process holds l now. +func (a *actor) holds(l *lease.Lease) bool { + a.mu.Lock() + defer a.mu.Unlock() + return a.held == l +} + // took is the lease taken: recorded, earlier epochs nobody gave back ended as expired, kept. func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Inventory, epoch uint64, gone chan struct{}) { a.mu.Lock() @@ -288,15 +320,19 @@ func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Invento h := holderOf(instance) recording, cancel := context.WithTimeout(ctx, 10*time.Second) expired, err := inv.TookEpoch(recording, inventory.Epoch{Epoch: epoch, Instance: h.Instance, Host: h.Host, - Build: h.Build, Taken: time.Now()}) + Build: h.Build, Taken: time.Now()}, l.GivenBack()) cancel() if err != nil { fmt.Printf("epoch %d could not be recorded as taken, so a stale refusal from it will not name it: %v\n", epoch, err) } + if given := l.GivenBack(); given != 0 { + fmt.Printf("the lease was given back by the controller of epoch %d: taken at once, at epoch %d\n", given, epoch) + } for _, e := range expired { fmt.Printf("the controller of epoch %d (%s) stopped renewing the lease without giving it back: it is "+ "taken over at epoch %d\n", e.Epoch, e.Instance, epoch) } + a.sayTakeover(ctx, epoch) go l.Keep(ctx) go func() { <-l.Lost() diff --git a/cmd/mesh-controller/lease_integration_test.go b/cmd/mesh-controller/lease_integration_test.go index 4ec880ec..9d38cadd 100644 --- a/cmd/mesh-controller/lease_integration_test.go +++ b/cmd/mesh-controller/lease_integration_test.go @@ -89,15 +89,28 @@ func TestTwoControllersOneActs(t *testing.T) { t.Fatalf("B, waiting, may act: %v", err) } - // A hands over, as a controller being replaced does: B takes the lease at once, at a higher epoch. + // A is stopped, as its service manager stops a controller being replaced (novox/hq issue 906): the stop + // alone gives the lease back, before anything else of A is taken down, and B takes it within a second, at + // a higher epoch — no wait for the key's age, and no condition said. stopA() - a.release() + stopped := time.Now() var lostB <-chan struct{} select { case lostB = <-tookB: case <-time.After(10 * time.Second): t.Fatal("B did not take the lease A gave back") } + if waited := time.Since(stopped); waited > time.Second { + t.Fatalf("B took the lease %s after A was stopped: the stop did not give it back first", waited) + } + a.release() // what serve's own end does after: waits for the release under way, gives nothing twice + conditionsOfB := conditions.NewInMemory() + keeperOfB := conditions.NewKeeper(ctx, conditions.Options{Store: conditionsOfB, History: conditionsOfB}) + b.watchTakeovers(ctx, inv, keeperOfB) + keeperOfB.Close(ctx) + if said, err := conditionsOfB.Since(ctx, stopped.Add(-time.Minute)); err != nil || len(said) != 0 { + t.Fatalf("a lease given back is said as a condition: %+v, %v", said, err) + } select { case <-lostA: default: diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 70c96fc7..3bbc8401 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -180,6 +180,9 @@ func serve(ctx context.Context) (err error) { if keeper != nil { defer closeConditions(keeper) defer stopRunning() // on an error return, the goroutines end before the keeper closes + // The takeover of the lease said as it happened, raised and cleared (novox/hq issue 906). + theLease.watchTakeovers(ctx, inv, keeper) + defer theLease.watchTakeovers(ctx, nil, nil) // the keeper closes after this: nothing said into it then } // The bus's own objects, asserted on every start. **Not created once at genesis**: a stream diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 18bfea30..3af94c2d 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "sort" "strings" @@ -147,9 +148,10 @@ var signalsTable = []signalRow{ return newestOf(f.machines, func(m machineFacts) time.Time { return m.toolsHeard }) }}, {Row: "S12", Signal: "the controller lease renewed", Emitter: "controller", Trigger: "every 5 s", - Bound: "15 s (the key's age); a holder that lost the lease, or stopped renewing and was taken over, and a " + - "lease bucket found raised again from nothing, are said for an hour after; a controller serving without " + - "the lease, for as long as it does", + Bound: "15 s (the key's age); a holder that lost the lease or stopped renewing it is said once by the " + + "controller that took it over, raised and cleared as it takes it, and for an hour after while none " + + "holds it; a lease given back is never said; a lease bucket found raised again from nothing is said " + + "for an hour after; a controller serving without the lease, for as long as it does", Kind: "lease-lost", Severity: conditions.Urgent, Phase: 2, needs: func(f *signalFacts) error { return f.leaseErr }, watch: watchLease, newest: func(f *signalFacts) time.Time { return f.lease.renewed }}, @@ -644,10 +646,29 @@ func watchLease(f *signalFacts) []conditions.Observation { "does not know it lost it", Said: l.resetSaid}) } + // An epoch below the one this controller holds was taken over by it: said once by the takeover itself, + // raised and cleared (sayTakeover), not observed for an hour after (novox/hq issue 906). One that ended + // while this controller does not hold the lease — serving without it — is still said for the hour. + var ended []inventory.Epoch + for _, e := range l.ended { + if l.held && e.Epoch < l.epoch { + continue + } + ended = append(ended, e) + } + if o, ok := lostObservation(ended, l.epoch, f.now); ok { + out = append(out, o) + } + return out +} + +// lostObservation is the lease-lost condition of the epochs that lost the lease or let it expire within the +// hour, named by the newest; false when none did. An epoch given back is no loss. +func lostObservation(ended []inventory.Epoch, acting uint64, now time.Time) (conditions.Observation, bool) { var lost []string newest := inventory.Epoch{} - for _, e := range l.ended { - if e.Ended == nil || e.How == inventory.EpochReleased || f.now.Sub(*e.Ended) > advisoryQuiet { + for _, e := range ended { + if e.Ended == nil || e.How == inventory.EpochReleased || now.Sub(*e.Ended) > advisoryQuiet { continue } lost = append(lost, fmt.Sprintf("epoch %d (%s) %s at %s", e.Epoch, e.Instance, e.How, @@ -656,18 +677,95 @@ func watchLease(f *signalFacts) []conditions.Observation { newest = e } } - if len(lost) > 0 { - how := "lost it: another holder wrote its key, or no renewal was made within the key's age" - if newest.How == inventory.EpochExpired { - how = "stopped renewing it without giving it back, and was taken over" + if len(lost) == 0 { + return conditions.Observation{}, false + } + how := "lost it: another holder wrote its key, or no renewal was made within the key's age" + if newest.How == inventory.EpochExpired { + how = "stopped renewing it without giving it back, and was taken over" + } + return conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost", + Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent, Source: "S12", + Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch, + newest.Instance, how, acting), + Said: strings.Join(lost, "; ")}, true +} + +// takeover is the condition a takeover says: the epoch just before the one taken, when it lost the lease or +// let it expire; false when it gave it back, or none is recorded. +func takeover(ended []inventory.Epoch, taken uint64, now time.Time) (conditions.Observation, bool) { + var before *inventory.Epoch + for i, e := range ended { + if e.Epoch < taken && (before == nil || e.Epoch > before.Epoch) { + before = &ended[i] } - out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost", - Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent, - Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch, - newest.Instance, how, l.epoch), - Said: strings.Join(lost, "; ")}) + } + if before == nil { + return conditions.Observation{}, false + } + return lostObservation([]inventory.Epoch{*before}, taken, now) +} + +// sayTakeover is the takeover said as an event (novox/hq issue 906): when the controller before this one lost +// the lease or let it expire, the lease-lost condition is raised and cleared at once by this controller, which +// holds the lease — the operator reads that it happened and that it is over, not an urgent alarm for an hour. +// A lease given back says nothing. Said once the condition store is open; a take before that is said then. +func (a *actor) sayTakeover(ctx context.Context, epoch uint64) { + a.mu.Lock() + keeper, inv := a.keeper, a.inv + a.mu.Unlock() + if keeper == nil || inv == nil || epoch == 0 { + return + } + now := time.Now() + reading, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + ended, err := inv.EpochsSince(reading, now.Add(-advisoryQuiet)) + if err != nil { + // Failure is loud: S12 no longer observes an epoch below the one held, so an unread takeover is said as + // one that was not a handover — raised, and left for S12 to clear (ADR 0558 §3). + fmt.Printf("how the epoch before %d ended cannot be read, so its takeover is said as an expiry: %v\n", epoch, err) + o := conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost", + Kind: "lease-lost", Severity: conditions.Urgent, Source: "S12", + Summary: fmt.Sprintf("the controller of epoch %d took the lease, and how the controller before it ended "+ + "cannot be read: it is said as one that stopped renewing without giving it back", epoch), + Said: err.Error()} + if _, err := keeper.Observe(reading, o); err != nil { + fmt.Printf("the takeover at epoch %d could not be said: %v\n", epoch, err) + } + return + } + sayTakeoverOf(reading, keeper, ended, epoch, now) +} + +// sayTakeoverOf raises and clears the takeover's condition over the epochs read. +func sayTakeoverOf(ctx context.Context, keeper *conditions.Keeper, ended []inventory.Epoch, epoch uint64, now time.Time) { + o, ok := takeover(ended, epoch, now) + if !ok { + return + } + if _, err := keeper.Observe(ctx, o); err != nil { + fmt.Printf("the takeover at epoch %d could not be said: %v\n", epoch, err) + return + } + if _, err := keeper.Clear(ctx, o.Key(), fmt.Sprintf("taken over: the controller of epoch %d holds the lease", epoch)); err != nil { + fmt.Printf("the takeover at epoch %d was said and could not be cleared: %v\n", epoch, err) + } +} + +// watchTakeovers gives the serving controller's condition store to the lease, and says the takeover of the +// epoch it holds, if it took one over before the store opened. +func (a *actor) watchTakeovers(ctx context.Context, inv *inventory.Inventory, keeper *conditions.Keeper) { + a.mu.Lock() + a.keeper, a.inv = keeper, inv + held := a.held + a.mu.Unlock() + if held == nil { + return + } + if epoch, err := held.Epoch(); err == nil { + a.sayTakeover(ctx, epoch) } - return out } // handActsWithin is how far back a repeated cause counts (S15). diff --git a/cmd/mesh-controller/signals_test.go b/cmd/mesh-controller/signals_test.go index 673a0987..acb554cf 100644 --- a/cmd/mesh-controller/signals_test.go +++ b/cmd/mesh-controller/signals_test.go @@ -491,13 +491,41 @@ func TestAStaleWriterIsNamedByItsEpoch(t *testing.T) { } } -// **The lease lost is said for an hour, and serving without it for as long as it lasts** (S12). +// **The lease lost is said for an hour while nobody took it over, a takeover is said once as it happens — +// raised and cleared by the controller that took it — a lease given back is never said, and serving without +// it for as long as it lasts** (S12, novox/hq issue 906). func TestALeaseLostOrMissingIsSaid(t *testing.T) { now := time.Now() f := calm(now) expired := now.Add(-59 * time.Minute) f.lease.ended = []inventory.Epoch{{Epoch: 41, Instance: "controller@anchor pid 7", Host: "anchor", Ended: &expired, How: inventory.EpochExpired}} + // Held at epoch 57, above it: the takeover said it, and no hour of observation follows. + if got := watchLease(f); len(got) != 0 { + t.Fatalf("an epoch this controller took over is observed after the takeover: %+v", got) + } + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Now: func() time.Time { return now }}) + sayTakeoverOf(t.Context(), k, f.lease.ended, 57, now) + if open, err := k.Open(t.Context()); err != nil || len(open) != 0 { + t.Fatalf("the takeover is left open: %+v, %v", open, err) + } + k.Close(t.Context()) // its transitions written to the history + history, err := store.Since(t.Context(), now.Add(-time.Minute)) + if err != nil { + t.Fatal(err) + } + var changes []string + for _, e := range history { + if e.Key == "core.controller.lease.lost" { + changes = append(changes, e.Change) + } + } + if !slices.Equal(changes, []string{"raised", "cleared"}) { + t.Fatalf("the takeover of an expired lease is not raised and cleared: %v", changes) + } + // Not holding the lease, the loss is said for the hour. + f.lease.held = false got := watchLease(f) if len(got) != 1 || got[0].Key() != "core.controller.lease.lost" || !strings.Contains(got[0].Summary, "epoch 41") || got[0].Severity != conditions.Urgent { @@ -512,6 +540,9 @@ func TestALeaseLostOrMissingIsSaid(t *testing.T) { if got := watchLease(f); len(got) != 0 { t.Fatalf("a lease given back is said as lost: %+v", got) } + if _, said := takeover(f.lease.ended, 57, now); said { + t.Fatal("the takeover of a lease given back is said") + } f.lease = leaseFacts{held: true, epoch: 501, renewed: now, reset: now.Add(-time.Minute), resetSaid: "moved past 500"} if got := watchLease(f); len(got) != 1 || got[0].Key() != "core.controller.lease.reset" { t.Fatalf("a lease bucket raised again from nothing was not said: %+v", got) diff --git a/internal/inventory/order.go b/internal/inventory/order.go index 9e15fa27..2dfec2ac 100644 --- a/internal/inventory/order.go +++ b/internal/inventory/order.go @@ -67,14 +67,23 @@ func (i *Inventory) HighestEpoch(ctx context.Context) (uint64, error) { } // TookEpoch records an epoch taken, and ends every earlier one still open as found expired: the lease -// was free to take, so whoever held it last neither holds it nor said it let go. Answers the epochs it -// ended that way, newest first. -func (i *Inventory) TookEpoch(ctx context.Context, e Epoch) ([]Epoch, error) { +// was free to take, so whoever held it last neither holds it nor said it let go. givenBack is the epoch the +// bus says gave the key back (the lease's own delete marker names it, novox/hq issue 906, ADR 0558 §3): that +// epoch, if still open, is ended as released — its holder gave the key back and had not yet recorded it — and +// every other as expired; zero is none. Answers the epochs it ended as expired, newest first. +func (i *Inventory) TookEpoch(ctx context.Context, e Epoch, givenBack uint64) ([]Epoch, error) { tx, err := i.store.Pool().Begin(ctx) if err != nil { return nil, err } defer func() { _ = tx.Rollback(ctx) }() + if givenBack != 0 && givenBack < e.Epoch { + if _, err := tx.Exec(ctx, + `update controller_epoch set ended = now(), how = $2 where epoch = $1 and ended is null`, + int64(givenBack), EpochReleased); err != nil { + return nil, err + } + } rows, err := tx.Query(ctx, `update controller_epoch set ended = now(), how = $2 where ended is null and epoch < $1 diff --git a/internal/lease/lease.go b/internal/lease/lease.go index a2f05265..1c1b5dda 100644 --- a/internal/lease/lease.go +++ b/internal/lease/lease.go @@ -34,9 +34,11 @@ import ( "encoding/json" "errors" "fmt" + "strconv" "sync" "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -115,6 +117,7 @@ type Options struct { // Lease is one instance's hold, or its wait for one. type Lease struct { kv jetstream.KeyValue + js jetstream.JetStream stream jetstream.Stream ttl time.Duration o Options @@ -139,6 +142,9 @@ type Lease struct { lostWhy error lost chan struct{} taken time.Time + // givenBack is the epoch whose holder gave back the key this instance took — the epoch its delete marker + // names on the bus — and zero when the key was left to expire, or freed by anything else (novox/hq issue 906). + givenBack uint64 // update writes the key at a revision: the bucket's own, or a test's bus that does not answer. update func(ctx context.Context, key string, value []byte, last uint64) (uint64, error) } @@ -184,7 +190,7 @@ func Open(ctx context.Context, js jetstream.JetStream, bucket string, o Options) return nil, fmt.Errorf("a lease renewed every %s with a margin of %s does not fit a key that lives %s", o.RenewEvery, o.Margin, status.TTL()) } - return &Lease{kv: kv, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil + return &Lease{kv: kv, js: js, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil } // TTL is how long the key lives unrenewed. @@ -271,6 +277,7 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) { } else if found { return 0, &heldBy{h} } + givenBack := l.wasGivenBack(ctx) if err := l.aboveTheFloor(ctx); err != nil { return 0, err } @@ -307,13 +314,51 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) { "back to be taken again", revision, floor) } l.mu.Lock() - l.held, l.epoch, l.revision, l.taken = true, revision, revision, now + l.held, l.epoch, l.revision, l.taken, l.givenBack = true, revision, revision, now, givenBack l.renewedAt(anchor) l.mu.Unlock() l.o.Say("took the controller lease at epoch %d (%s)", revision, h.Instance) return revision, nil } +// ReleasedHeader is the header of the delete marker a holder's Release writes, naming the epoch it gave back: +// the one mark that says a handover. A delete marker without it — a command's own lease given back, a take +// given back over the floor — names no epoch, so it says nothing about the holder before (novox/hq issue 906). +const ReleasedHeader = "Mesh-Lease-Released" + +// wasGivenBack is the epoch the free key was given back by: the epoch its last message on the bus — the delete +// marker a holder's Release writes — names. Zero when the key expired (the bucket's age removed every message), +// when the marker names no epoch, and when it cannot be read: a takeover nobody can prove was a handover is +// said as one that was not (novox/hq issue 906, ADR 0558 §3). +func (l *Lease) wasGivenBack(ctx context.Context) uint64 { + if l.stream == nil { + return 0 + } + last, err := l.stream.GetLastMsgForSubject(ctx, l.subject()) + if err != nil || last == nil { + return 0 + } + if op := last.Header.Get("KV-Operation"); op != "DEL" && op != "PURGE" { + return 0 + } + epoch, err := strconv.ParseUint(last.Header.Get(ReleasedHeader), 10, 64) + if err != nil { + return 0 + } + return epoch +} + +// subject is the lease key's subject in its bucket's stream. +func (l *Lease) subject() string { return "$KV." + l.kv.Bucket() + "." + Key } + +// GivenBack is the epoch whose holder gave back the key this instance took, zero when it was not given back: +// what the takeover was, read from the bus (novox/hq issue 906). +func (l *Lease) GivenBack() uint64 { + l.mu.Lock() + defer l.mu.Unlock() + return l.givenBack +} + // floor is the highest epoch issued, zero without a floor. func (l *Lease) floor(ctx context.Context) (uint64, error) { if l.o.Floor == nil { @@ -588,9 +633,11 @@ func (l *Lease) lose(why error) error { return l.lostWhy } -// Release gives the lease back, so the next candidate takes it at once rather than after its age. -// Only the key this instance last wrote is deleted: one that moved is somebody else's. -func (l *Lease) Release(ctx context.Context) { +// Release gives the lease back, so the next candidate takes it at once rather than after its age, and answers +// the epoch it gave back — zero when it gave nothing back: it held none, had lost it, or the bus would not +// delete the key. Only the key this instance last wrote is deleted: one that moved is somebody else's. The +// gate closes before the key goes, so nothing this instance starts after overlaps its successor. +func (l *Lease) Release(ctx context.Context) uint64 { l.mu.Lock() held, revision, epoch := l.held && l.lostWhy == nil, l.revision, l.epoch l.held = false @@ -600,11 +647,11 @@ func (l *Lease) Release(ctx context.Context) { } l.mu.Unlock() if !held { - return + return 0 } releasing, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) defer cancel() - err := l.kv.Delete(releasing, Key, jetstream.LastRevision(revision)) + err := l.giveBack(releasing, epoch, revision) if err != nil && isWrongLast(err) { // A renewal was in flight as this began and landed first: the key moved, and is still this // instance's if it still names it at this epoch. Deleted at the revision it has now. @@ -612,14 +659,29 @@ func (l *Lease) Release(ctx context.Context) { var h Holder if json.Unmarshal(entry.Value(), &h) == nil && h.Instance == l.o.Holder.Instance && (h.Epoch == epoch || h.Epoch == 0 && entry.Revision() == epoch) { - err = l.kv.Delete(releasing, Key, jetstream.LastRevision(entry.Revision())) + err = l.giveBack(releasing, epoch, entry.Revision()) } } } if err != nil { l.o.Say("the controller lease (epoch %d) could not be given back, so the next controller waits for "+ "it to expire: %v", epoch, err) - return + return 0 } l.o.Say("gave the controller lease back (epoch %d)", epoch) + return epoch +} + +// giveBack deletes the key at the revision this instance last wrote, as the bucket's own delete does — a +// marker with KV-Operation DEL, compare-and-set on the key's last revision — and names the epoch given back in +// the marker, so the next holder can tell this handover from any other delete (ReleasedHeader). +func (l *Lease) giveBack(ctx context.Context, epoch, revision uint64) error { + if l.js == nil { + return l.kv.Delete(ctx, Key, jetstream.LastRevision(revision)) + } + msg := nats.NewMsg(l.subject()) + msg.Header.Set("KV-Operation", "DEL") + msg.Header.Set(ReleasedHeader, strconv.FormatUint(epoch, 10)) + _, err := l.js.PublishMsg(ctx, msg, jetstream.WithExpectLastSequencePerSubject(revision)) + return err } diff --git a/internal/lease/lease_test.go b/internal/lease/lease_test.go index e2fcf4d9..56267990 100644 --- a/internal/lease/lease_test.go +++ b/internal/lease/lease_test.go @@ -133,7 +133,12 @@ func TestTheSecondControllerWaitsAndTakesAHigherEpochOnHandover(t *testing.T) { // a gives it back: b takes it at once, not after the key's age, at a higher epoch. stop() handedOver := time.Now() - a.Release(t.Context()) + if given := a.Release(t.Context()); given != epochA { + t.Fatalf("a's release answers epoch %d given back, want %d", given, epochA) + } + if again := a.Release(t.Context()); again != 0 { + t.Fatalf("a second release gives epoch %d back again", again) + } select { case epochB := <-took: if epochB <= epochA { @@ -142,6 +147,9 @@ func TestTheSecondControllerWaitsAndTakesAHigherEpochOnHandover(t *testing.T) { if waited := time.Since(handedOver); waited > testTTL { t.Fatalf("b took the lease %s after a gave it back: it waited out the key's age", waited) } + if given := b.GivenBack(); given != epochA { + t.Fatalf("b reads from the bus that epoch %d gave the key back, want a's %d (novox/hq issue 906)", given, epochA) + } case <-time.After(2 * testTTL): t.Fatal("b never took the lease a gave back") } @@ -182,6 +190,9 @@ func TestAHolderThatStopsRenewingStopsActingBeforeItIsTakenOver(t *testing.T) { if epochB <= epochA { t.Fatalf("b took epoch %d after a's %d", epochB, epochA) } + if given := b.GivenBack(); given != 0 { + t.Fatalf("b reads a key that expired as given back by epoch %d (novox/hq issue 906)", given) + } // And a, the moment it tries to renew, knows it lost: the key moved. if err := a.Renew(t.Context()); err == nil { t.Fatal("a renewed a lease b holds") @@ -226,8 +237,11 @@ func TestARenewalRefusedIsALossAtOnce(t *testing.T) { if _, err := a.Epoch(); !errors.Is(err, ErrNotHeld) { t.Fatalf("a acts after its renewal was refused: %v", err) } - // And giving back a lease it lost deletes nothing of the one who holds it now. - a.Release(t.Context()) + // And giving back a lease it lost deletes nothing of the one who holds it now, and answers no epoch given + // back, so none is recorded released (novox/hq issue 906). + if given := a.Release(t.Context()); given != 0 { + t.Fatalf("a says it gave back epoch %d of a lease it lost", given) + } if h, found, err := Current(t.Context(), kv); err != nil || !found || h.Instance != "intruder" { t.Fatalf("after a gave back what it lost, the key holds %+v (%v, %v)", h, found, err) }