diff --git a/cmd/mesh-controller/acting.go b/cmd/mesh-controller/acting.go index 637014f6..ebf60696 100644 --- a/cmd/mesh-controller/acting.go +++ b/cmd/mesh-controller/acting.go @@ -325,8 +325,8 @@ func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Invento 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) + 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 "+ diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index d1144ac1..3af94c2d 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -722,7 +722,17 @@ func (a *actor) sayTakeover(ctx context.Context, epoch uint64) { 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) + // 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) diff --git a/internal/inventory/order.go b/internal/inventory/order.go index 35a55d88..2dfec2ac 100644 --- a/internal/inventory/order.go +++ b/internal/inventory/order.go @@ -67,21 +67,20 @@ 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. 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) { +// 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 { + if givenBack != 0 && givenBack < e.Epoch { 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 { + `update controller_epoch set ended = now(), how = $2 where epoch = $1 and ended is null`, + int64(givenBack), EpochReleased); err != nil { return nil, err } } diff --git a/internal/lease/lease.go b/internal/lease/lease.go index a4fb3b30..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,9 +142,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 + // 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) } @@ -187,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. @@ -318,25 +321,39 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) { 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 { +// 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 false + return 0 } - last, err := l.stream.GetLastMsgForSubject(ctx, "$KV."+l.kv.Bucket()+"."+Key) + last, err := l.stream.GetLastMsgForSubject(ctx, l.subject()) if err != nil || last == nil { - return false + return 0 } - op := last.Header.Get("KV-Operation") - return op == "DEL" || op == "PURGE" + 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 } -// GivenBack is whether the key this instance took had been given back by its last holder, not left to expire: +// 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() bool { +func (l *Lease) GivenBack() uint64 { l.mu.Lock() defer l.mu.Unlock() return l.givenBack @@ -634,7 +651,7 @@ func (l *Lease) Release(ctx context.Context) uint64 { } 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. @@ -642,7 +659,7 @@ func (l *Lease) Release(ctx context.Context) uint64 { 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()) } } } @@ -654,3 +671,17 @@ func (l *Lease) Release(ctx context.Context) uint64 { 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 56b3961a..56267990 100644 --- a/internal/lease/lease_test.go +++ b/internal/lease/lease_test.go @@ -147,8 +147,8 @@ 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)") + 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") @@ -190,8 +190,8 @@ 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)") + 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 {