diff --git a/cmd/mesh-controller/acting.go b/cmd/mesh-controller/acting.go index fe5aca76..637014f6 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 l.GivenBack() { + fmt.Printf("the lease was given back by the controller before this one: taken at once, at epoch %d\n", 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..d1144ac1 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,85 @@ 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 { + fmt.Printf("how the epoch before %d ended cannot be read, so its takeover is not 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..35a55d88 100644 --- a/internal/inventory/order.go +++ b/internal/inventory/order.go @@ -67,14 +67,24 @@ 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 bus's word +// that the key was given back (the lease's own delete marker, novox/hq issue 906): then the newest earlier +// epoch still open is ended as released — its holder gave the key back and had not yet recorded it — and only +// the ones before it as expired. Answers the epochs it ended as expired, newest first. +func (i *Inventory) TookEpoch(ctx context.Context, e Epoch, givenBack bool) ([]Epoch, error) { tx, err := i.store.Pool().Begin(ctx) if err != nil { return nil, err } defer func() { _ = tx.Rollback(ctx) }() + if givenBack { + if _, err := tx.Exec(ctx, + `update controller_epoch set ended = now(), how = $2 + where epoch = (select max(epoch) from controller_epoch where ended is null and epoch < $1)`, + int64(e.Epoch), 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..a4fb3b30 100644 --- a/internal/lease/lease.go +++ b/internal/lease/lease.go @@ -139,6 +139,9 @@ type Lease struct { lostWhy error lost chan struct{} taken time.Time + // givenBack is whether the key this instance took had been given back by its last holder — the bus's + // delete marker on it — rather than left to expire (novox/hq issue 906). + givenBack bool // 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) } @@ -271,6 +274,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 +311,37 @@ 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 } +// wasGivenBack is whether the free key was given back: its last message on the bus is the delete marker a +// holder's Release writes. A key that expired has no message left — the bucket's age removed it — and one that +// cannot be read is answered as not given back, so a takeover nobody can prove was a handover is said as one +// that was not (novox/hq issue 906). +func (l *Lease) wasGivenBack(ctx context.Context) bool { + if l.stream == nil { + return false + } + last, err := l.stream.GetLastMsgForSubject(ctx, "$KV."+l.kv.Bucket()+"."+Key) + if err != nil || last == nil { + return false + } + op := last.Header.Get("KV-Operation") + return op == "DEL" || op == "PURGE" +} + +// GivenBack is whether the key this instance took had been given back by its last holder, not left to expire: +// what the takeover was, read from the bus (novox/hq issue 906). +func (l *Lease) GivenBack() bool { + 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 +616,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,7 +630,7 @@ 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() @@ -619,7 +649,8 @@ func (l *Lease) Release(ctx context.Context) { 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 } diff --git a/internal/lease/lease_test.go b/internal/lease/lease_test.go index e2fcf4d9..56b3961a 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 !b.GivenBack() { + t.Fatal("b does not read from the bus that a gave the key back (novox/hq issue 906)") + } 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 b.GivenBack() { + t.Fatal("b reads a key that expired as given back (novox/hq issue 906)") + } // 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) }