Name the epoch given back in the lease's delete marker, so only a holder's own release counts as a handover (issue 906, ADR 0558 §3, review round 1)

A plain delete marker on the lease key is also written by a command's own
lease and by a take given back over the floor; counted as a handover, it
would record a crashed holder's epoch released and say nothing. Release now
writes the marker itself, compare-and-set on the key's revision, with the
header Mesh-Lease-Released naming its epoch, and the next holder ends only
that epoch as released. A takeover whose predecessor cannot be read is now
raised as an expiry, not only printed.

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() {
+	if given := l.GivenBack(); given != 0 {
-		fmt.Printf("the lease was given back by the controller before this one: taken at once, at epoch %d\n", epoch)
+		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 "+

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)

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
+// was free to take, so whoever held it last neither holds it nor said it let go. givenBack is the epoch the
-// that the key was given back (the lease's own delete marker, novox/hq issue 906): then the newest earlier
+// bus says gave the key back (the lease's own delete marker names it, novox/hq issue 906, ADR 0558 §3): that
-// epoch still open is ended as released — its holder gave the key back and had not yet recorded it — and only
+// epoch, if still open, is ended as released — its holder gave the key back and had not yet recorded it — and
-// the ones before it as expired. Answers the epochs it ended as expired, newest first.
+// 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 bool) ([]Epoch, error) {
+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
+			`update controller_epoch set ended = now(), how = $2 where epoch = $1 and ended is null`,
-			  where epoch = (select max(epoch) from controller_epoch where ended is null and epoch < $1)`,
+			int64(givenBack), EpochReleased); err != nil {
-			int64(e.Epoch), EpochReleased); err != nil {
 			return nil, err
 		}
 	}

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
+	// givenBack is the epoch whose holder gave back the key this instance took — the epoch its delete marker
-	// delete marker on it — rather than left to expire (novox/hq issue 906).
+	// names on the bus — and zero when the key was left to expire, or freed by anything else (novox/hq issue 906).
-	givenBack bool
+	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
+// ReleasedHeader is the header of the delete marker a holder's Release writes, naming the epoch it gave back:
-// holder's Release writes. A key that expired has no message left — the bucket's age removed it — and one that
+// the one mark that says a handover. A delete marker without it — a command's own lease given back, a take
-// cannot be read is answered as not given back, so a takeover nobody can prove was a handover is said as one
+// given back over the floor — names no epoch, so it says nothing about the holder before (novox/hq issue 906).
-// that was not (novox/hq issue 906).
+const ReleasedHeader = "Mesh-Lease-Released"
-func (l *Lease) wasGivenBack(ctx context.Context) bool {
+
+// 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")
+	if op := last.Header.Get("KV-Operation"); op != "DEL" && op != "PURGE" {
-	return 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
+}

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() {
+		if given := b.GivenBack(); given != epochA {
-			t.Fatal("b does not read from the bus that a gave the key back (novox/hq issue 906)")
+			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() {
+		if given := b.GivenBack(); given != 0 {
-			t.Fatal("b reads a key that expired as given back (novox/hq issue 906)")
+			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 {