diff --git a/kv/active_timestamp_tracker.go b/kv/active_timestamp_tracker.go
index 18f387695..c121eafab 100644
--- a/kv/active_timestamp_tracker.go
+++ b/kv/active_timestamp_tracker.go
@@ -1,13 +1,85 @@
 package kv
 
-import "sync"
+import (
+	"encoding/hex"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/cockroachdb/errors"
+)
+
+const (
+	defaultMaxActiveBackupPins = 64
+	defaultBackupPinSweepEvery = time.Second
+)
+
+var (
+	ErrInvalidBackupPin     = errors.New("backup pin is invalid")
+	ErrTooManyActiveBackups = errors.New("too many active backup pins")
+)
+
+type BackupPinID [16]byte
+
+func (id BackupPinID) IsZero() bool {
+	return id == BackupPinID{}
+}
+
+func (id BackupPinID) String() string {
+	return hex.EncodeToString(id[:])
+}
+
+type backupDeadlinePin struct {
+	readTS   uint64
+	deadline time.Time
+}
+
+type backupPinKey struct {
+	id      BackupPinID
+	groupID uint64
+}
+
+func newBackupPinKey(pinID BackupPinID, groupID uint64) backupPinKey {
+	return backupPinKey{id: pinID, groupID: groupID}
+}
+
+type ActiveTimestampTrackerOption func(*ActiveTimestampTracker)
+
+func WithActiveTimestampTrackerMaxBackupPins(maxPins int) ActiveTimestampTrackerOption {
+	return func(t *ActiveTimestampTracker) {
+		if maxPins > 0 {
+			t.maxBackupPins = maxPins
+		}
+	}
+}
+
+func WithActiveTimestampTrackerSweepInterval(interval time.Duration) ActiveTimestampTrackerOption {
+	return func(t *ActiveTimestampTracker) {
+		t.sweepEvery = interval
+	}
+}
+
+func WithActiveTimestampTrackerLogger(logger *slog.Logger) ActiveTimestampTrackerOption {
+	return func(t *ActiveTimestampTracker) {
+		if logger != nil {
+			t.logger = logger
+		}
+	}
+}
 
 // ActiveTimestampTracker tracks in-flight read or transaction timestamps that
 // must remain readable while background compaction is running.
 type ActiveTimestampTracker struct {
-	mu     sync.Mutex
-	nextID uint64
-	active map[uint64]uint64
+	mu            sync.Mutex
+	nextID        uint64
+	active        map[uint64]uint64
+	backupPins    map[backupPinKey]backupDeadlinePin
+	maxBackupPins int
+	sweepEvery    time.Duration
+	sweepOnce     sync.Once
+	stopCh        chan struct{}
+	closeOnce     sync.Once
+	logger        *slog.Logger
 }
 
 // ActiveTimestampToken releases one tracked timestamp when the owning
@@ -18,10 +90,21 @@ type ActiveTimestampToken struct {
 	once    sync.Once
 }
 
-func NewActiveTimestampTracker() *ActiveTimestampTracker {
-	return &ActiveTimestampTracker{
-		active: make(map[uint64]uint64),
+func NewActiveTimestampTracker(opts ...ActiveTimestampTrackerOption) *ActiveTimestampTracker {
+	t := &ActiveTimestampTracker{
+		active:        make(map[uint64]uint64),
+		backupPins:    make(map[backupPinKey]backupDeadlinePin),
+		maxBackupPins: defaultMaxActiveBackupPins,
+		sweepEvery:    defaultBackupPinSweepEvery,
+		stopCh:        make(chan struct{}),
+		logger:        slog.Default(),
+	}
+	for _, opt := range opts {
+		if opt != nil {
+			opt(t)
+		}
 	}
+	return t
 }
 
 func (t *ActiveTimestampTracker) Pin(ts uint64) *ActiveTimestampToken {
@@ -51,9 +134,217 @@ func (t *ActiveTimestampTracker) Oldest() uint64 {
 			oldest = ts
 		}
 	}
+	now := time.Now()
+	for _, pin := range t.backupPins {
+		if !pin.deadline.After(now) {
+			continue
+		}
+		if oldest == 0 || pin.readTS < oldest {
+			oldest = pin.readTS
+		}
+	}
 	return oldest
 }
 
+func (t *ActiveTimestampTracker) PinWithDeadline(pinID BackupPinID, readTS uint64, deadline time.Time) error {
+	return t.PinWithDeadlineForGroup(pinID, 0, readTS, deadline)
+}
+
+func (t *ActiveTimestampTracker) PinWithDeadlineForGroup(pinID BackupPinID, groupID uint64, readTS uint64, deadline time.Time) error {
+	return t.pinWithDeadlineForGroup(pinID, groupID, readTS, deadline, true)
+}
+
+func (t *ActiveTimestampTracker) ApplyPinWithDeadlineForGroup(pinID BackupPinID, groupID uint64, readTS uint64, deadline time.Time) error {
+	return t.pinWithDeadlineForGroup(pinID, groupID, readTS, deadline, false)
+}
+
+func (t *ActiveTimestampTracker) pinWithDeadlineForGroup(pinID BackupPinID, groupID uint64, readTS uint64, deadline time.Time, enforceLimit bool) error {
+	if t == nil {
+		return nil
+	}
+	if pinID.IsZero() || readTS == 0 || readTS == ^uint64(0) || deadline.IsZero() {
+		return errors.WithStack(ErrInvalidBackupPin)
+	}
+	t.mu.Lock()
+	expired := t.reapExpiredBackupPinsLocked(time.Now())
+	key := newBackupPinKey(pinID, groupID)
+	if enforceLimit && !t.hasBackupPinIDLocked(pinID) && t.activeBackupPinIDCountLocked() >= t.maxBackupPins {
+		t.mu.Unlock()
+		t.logExpiredBackupPins(expired)
+		return errors.WithStack(ErrTooManyActiveBackups)
+	}
+	t.backupPins[key] = backupDeadlinePin{readTS: readTS, deadline: deadline}
+	t.startBackupPinSweeperLocked()
+	t.mu.Unlock()
+	t.logExpiredBackupPins(expired)
+	return nil
+}
+
+func (t *ActiveTimestampTracker) Extend(pinID BackupPinID, deadline time.Time) error {
+	return t.ExtendForGroup(pinID, 0, deadline)
+}
+
+func (t *ActiveTimestampTracker) ExtendForGroup(pinID BackupPinID, groupID uint64, deadline time.Time) error {
+	return t.extendForGroup(pinID, groupID, deadline, true)
+}
+
+func (t *ActiveTimestampTracker) ApplyExtendForGroup(pinID BackupPinID, groupID uint64, deadline time.Time) error {
+	return t.extendForGroup(pinID, groupID, deadline, false)
+}
+
+func (t *ActiveTimestampTracker) extendForGroup(pinID BackupPinID, groupID uint64, deadline time.Time, returnMissingExpired bool) error {
+	if t == nil {
+		return nil
+	}
+	if pinID.IsZero() || deadline.IsZero() {
+		return errors.WithStack(ErrInvalidBackupPin)
+	}
+	t.mu.Lock()
+	key := newBackupPinKey(pinID, groupID)
+	pin, exists := t.backupPins[key]
+	if !exists {
+		t.mu.Unlock()
+		if !returnMissingExpired {
+			return nil
+		}
+		return errors.WithStack(ErrInvalidBackupPin)
+	}
+	if !pin.deadline.After(time.Now()) {
+		delete(t.backupPins, key)
+		t.mu.Unlock()
+		t.logExpiredBackupPins([]expiredBackupPin{{key: key, ts: pin.readTS}})
+		if !returnMissingExpired {
+			return nil
+		}
+		return errors.WithStack(ErrInvalidBackupPin)
+	}
+	pin.deadline = deadline
+	t.backupPins[key] = pin
+	t.mu.Unlock()
+	return nil
+}
+
+func (t *ActiveTimestampTracker) ReleaseBackupPin(pinID BackupPinID) {
+	t.ReleaseBackupPinForGroup(pinID, 0)
+}
+
+func (t *ActiveTimestampTracker) ReleaseBackupPinForGroup(pinID BackupPinID, groupID uint64) {
+	if t == nil || pinID.IsZero() {
+		return
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	delete(t.backupPins, newBackupPinKey(pinID, groupID))
+}
+
+func (t *ActiveTimestampTracker) ActiveBackupPinCount() int {
+	if t == nil {
+		return 0
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return len(t.backupPins)
+}
+
+func (t *ActiveTimestampTracker) BackupPinDeadline(pinID BackupPinID) (time.Time, bool) {
+	return t.BackupPinDeadlineForGroup(pinID, 0)
+}
+
+func (t *ActiveTimestampTracker) BackupPinDeadlineForGroup(pinID BackupPinID, groupID uint64) (time.Time, bool) {
+	if t == nil || pinID.IsZero() {
+		return time.Time{}, false
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	pin, ok := t.backupPins[newBackupPinKey(pinID, groupID)]
+	return pin.deadline, ok
+}
+
+func (t *ActiveTimestampTracker) startBackupPinSweeperLocked() {
+	if t.sweepEvery <= 0 {
+		return
+	}
+	t.sweepOnce.Do(func() {
+		go t.sweepBackupPins()
+	})
+}
+
+func (t *ActiveTimestampTracker) sweepBackupPins() {
+	if t.sweepEvery <= 0 {
+		return
+	}
+	ticker := time.NewTicker(t.sweepEvery)
+	defer ticker.Stop()
+	for {
+		select {
+		case now := <-ticker.C:
+			t.reapExpiredBackupPins(now)
+		case <-t.stopCh:
+			return
+		}
+	}
+}
+
+// Close stops the backup-pin sweeper goroutine. It is safe to call more than
+// once and on trackers that never started the sweeper.
+func (t *ActiveTimestampTracker) Close() {
+	if t == nil {
+		return
+	}
+	t.closeOnce.Do(func() {
+		close(t.stopCh)
+	})
+}
+
+func (t *ActiveTimestampTracker) reapExpiredBackupPins(now time.Time) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	expired := t.reapExpiredBackupPinsLocked(now)
+	t.mu.Unlock()
+	t.logExpiredBackupPins(expired)
+}
+
+type expiredBackupPin struct {
+	key backupPinKey
+	ts  uint64
+}
+
+func (t *ActiveTimestampTracker) reapExpiredBackupPinsLocked(now time.Time) []expiredBackupPin {
+	expired := make([]expiredBackupPin, 0)
+	for key, pin := range t.backupPins {
+		if !pin.deadline.After(now) {
+			expired = append(expired, expiredBackupPin{key: key, ts: pin.readTS})
+			delete(t.backupPins, key)
+		}
+	}
+	return expired
+}
+
+func (t *ActiveTimestampTracker) hasBackupPinIDLocked(pinID BackupPinID) bool {
+	for key := range t.backupPins {
+		if key.id == pinID {
+			return true
+		}
+	}
+	return false
+}
+
+func (t *ActiveTimestampTracker) activeBackupPinIDCountLocked() int {
+	seen := make(map[BackupPinID]struct{}, len(t.backupPins))
+	for key := range t.backupPins {
+		seen[key.id] = struct{}{}
+	}
+	return len(seen)
+}
+
+func (t *ActiveTimestampTracker) logExpiredBackupPins(expired []expiredBackupPin) {
+	for _, pin := range expired {
+		t.logger.Warn("backup_pin_expired", "pin_id", pin.key.id.String(), "raft_group_id", pin.key.groupID, "read_ts", pin.ts)
+	}
+}
+
 func (t *ActiveTimestampToken) Release() {
 	if t == nil || t.tracker == nil || t.id == 0 {
 		return
diff --git a/kv/active_timestamp_tracker_test.go b/kv/active_timestamp_tracker_test.go
index cd35f0648..81777b976 100644
--- a/kv/active_timestamp_tracker_test.go
+++ b/kv/active_timestamp_tracker_test.go
@@ -2,6 +2,7 @@ package kv
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -21,3 +22,169 @@ func TestActiveTimestampTrackerOldest(t *testing.T) {
 	second.Release()
 	require.Equal(t, uint64(30), tracker.Oldest())
 }
+
+func TestActiveTimestampTrackerOldestIncludesBackupPins(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+
+	token := tracker.Pin(30)
+	defer token.Release()
+
+	pinID := backupTrackerTestPinID(1)
+	require.NoError(t, tracker.PinWithDeadline(pinID, 20, time.Now().Add(time.Hour)))
+	require.Equal(t, uint64(20), tracker.Oldest())
+
+	tracker.ReleaseBackupPin(pinID)
+	require.Equal(t, uint64(30), tracker.Oldest())
+}
+
+func TestActiveTimestampTrackerBackupPinExpiry(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	now := time.UnixMilli(3000)
+
+	pinID := backupTrackerTestPinID(1)
+	require.NoError(t, tracker.PinWithDeadline(pinID, 20, now.Add(-time.Millisecond)))
+	require.Equal(t, uint64(0), tracker.Oldest())
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+
+	tracker.reapExpiredBackupPins(now)
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(0), tracker.Oldest())
+}
+
+func TestActiveTimestampTrackerBackupPinExtendMovesDeadline(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	pinID := backupTrackerTestPinID(1)
+	now := time.Now()
+	firstDeadline := now.Add(time.Hour)
+	secondDeadline := now.Add(2 * time.Hour)
+
+	require.NoError(t, tracker.PinWithDeadline(pinID, 20, firstDeadline))
+	require.NoError(t, tracker.Extend(pinID, secondDeadline))
+
+	got, ok := tracker.BackupPinDeadline(pinID)
+	require.True(t, ok)
+	require.Equal(t, secondDeadline, got)
+	tracker.reapExpiredBackupPins(firstDeadline.Add(time.Millisecond))
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+}
+
+func TestActiveTimestampTrackerBackupPinExtendMissingIsInvalid(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+
+	require.ErrorIs(t, tracker.Extend(backupTrackerTestPinID(1), time.UnixMilli(5000)), ErrInvalidBackupPin)
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+}
+
+func TestActiveTimestampTrackerBackupPinExtendExpiredIsInvalid(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	pinID := backupTrackerTestPinID(1)
+	now := time.Now()
+
+	require.NoError(t, tracker.PinWithDeadline(pinID, 20, now.Add(-time.Millisecond)))
+	require.ErrorIs(t, tracker.Extend(pinID, now.Add(time.Hour)), ErrInvalidBackupPin)
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(0), tracker.Oldest())
+}
+
+func TestActiveTimestampTrackerBackupPinReleaseIsIdempotent(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	pinID := backupTrackerTestPinID(1)
+
+	require.NoError(t, tracker.PinWithDeadline(pinID, 20, time.UnixMilli(5000)))
+	tracker.ReleaseBackupPin(pinID)
+	tracker.ReleaseBackupPin(pinID)
+
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(0), tracker.Oldest())
+}
+
+func TestActiveTimestampTrackerBackupPinLimit(t *testing.T) {
+	tracker := NewActiveTimestampTracker(
+		WithActiveTimestampTrackerSweepInterval(0),
+		WithActiveTimestampTrackerMaxBackupPins(1),
+	)
+	first := backupTrackerTestPinID(1)
+	second := backupTrackerTestPinID(2)
+
+	now := time.Now()
+	require.NoError(t, tracker.PinWithDeadline(first, 20, now.Add(time.Hour)))
+	require.NoError(t, tracker.PinWithDeadline(first, 25, now.Add(2*time.Hour)))
+	require.ErrorIs(t, tracker.PinWithDeadline(second, 30, now.Add(3*time.Hour)), ErrTooManyActiveBackups)
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(25), tracker.Oldest())
+}
+
+func TestActiveTimestampTrackerBackupPinLimitCountsLogicalPinIDs(t *testing.T) {
+	tracker := NewActiveTimestampTracker(
+		WithActiveTimestampTrackerSweepInterval(0),
+		WithActiveTimestampTrackerMaxBackupPins(1),
+	)
+	first := backupTrackerTestPinID(1)
+	second := backupTrackerTestPinID(2)
+	deadline := time.Now().Add(time.Hour)
+
+	require.NoError(t, tracker.PinWithDeadlineForGroup(first, 1, 20, deadline))
+	require.NoError(t, tracker.PinWithDeadlineForGroup(first, 2, 20, deadline))
+	require.Equal(t, 2, tracker.ActiveBackupPinCount())
+	require.ErrorIs(t, tracker.PinWithDeadlineForGroup(second, 3, 30, deadline), ErrTooManyActiveBackups)
+}
+
+func TestActiveTimestampTrackerBackupPinLimitReapsExpiredPinsFirst(t *testing.T) {
+	tracker := NewActiveTimestampTracker(
+		WithActiveTimestampTrackerSweepInterval(0),
+		WithActiveTimestampTrackerMaxBackupPins(1),
+	)
+	first := backupTrackerTestPinID(1)
+	second := backupTrackerTestPinID(2)
+	now := time.Now()
+
+	require.NoError(t, tracker.PinWithDeadline(first, 20, now.Add(-time.Millisecond)))
+	require.NoError(t, tracker.PinWithDeadline(second, 30, now.Add(time.Hour)))
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(30), tracker.Oldest())
+}
+
+func TestActiveTimestampTrackerBackupPinsAreScopedByRaftGroup(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	pinID := backupTrackerTestPinID(1)
+	deadline := time.Now().Add(time.Hour)
+
+	require.NoError(t, tracker.PinWithDeadlineForGroup(pinID, 1, 20, deadline))
+	require.NoError(t, tracker.PinWithDeadlineForGroup(pinID, 2, 20, deadline))
+	require.Equal(t, 2, tracker.ActiveBackupPinCount())
+
+	tracker.ReleaseBackupPinForGroup(pinID, 1)
+	_, ok := tracker.BackupPinDeadlineForGroup(pinID, 1)
+	require.False(t, ok)
+	got, ok := tracker.BackupPinDeadlineForGroup(pinID, 2)
+	require.True(t, ok)
+	require.Equal(t, deadline, got)
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+}
+
+func TestActiveTimestampTrackerRejectsInvalidBackupPins(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	validID := backupTrackerTestPinID(1)
+
+	require.ErrorIs(t, tracker.PinWithDeadline(BackupPinID{}, 20, time.UnixMilli(5000)), ErrInvalidBackupPin)
+	require.ErrorIs(t, tracker.PinWithDeadline(validID, 0, time.UnixMilli(5000)), ErrInvalidBackupPin)
+	require.ErrorIs(t, tracker.PinWithDeadline(validID, ^uint64(0), time.UnixMilli(5000)), ErrInvalidBackupPin)
+	require.ErrorIs(t, tracker.PinWithDeadline(validID, 20, time.Time{}), ErrInvalidBackupPin)
+	require.ErrorIs(t, tracker.Extend(BackupPinID{}, time.UnixMilli(5000)), ErrInvalidBackupPin)
+	require.ErrorIs(t, tracker.Extend(validID, time.Time{}), ErrInvalidBackupPin)
+}
+
+func TestActiveTimestampTrackerCloseIsIdempotent(t *testing.T) {
+	tracker := NewActiveTimestampTracker()
+
+	tracker.Close()
+	tracker.Close()
+}
+
+func backupTrackerTestPinID(seed byte) BackupPinID {
+	var id BackupPinID
+	for i := range id {
+		id[i] = seed + byte(i)
+	}
+	return id
+}
diff --git a/kv/backup_codec.go b/kv/backup_codec.go
new file mode 100644
index 000000000..ee18da163
--- /dev/null
+++ b/kv/backup_codec.go
@@ -0,0 +1,162 @@
+package kv
+
+import (
+	"encoding/binary"
+	"math"
+	"time"
+
+	"github.com/cockroachdb/errors"
+)
+
+const (
+	raftEncodeBackup byte = 0x0e
+
+	backupSubtypePin     byte = 0x01
+	backupSubtypeExtend  byte = 0x02
+	backupSubtypeRelease byte = 0x03
+
+	backupPinIDBytes = 16
+	backupUint64Size = 8
+
+	backupEnvelopeHeaderLen = 2
+	backupPinEntryLen       = backupEnvelopeHeaderLen + backupPinIDBytes + backupUint64Size + backupUint64Size
+	backupExtendEntryLen    = backupEnvelopeHeaderLen + backupPinIDBytes + backupUint64Size
+	backupReleaseEntryLen   = backupEnvelopeHeaderLen + backupPinIDBytes
+
+	backupPinIDStart        = backupEnvelopeHeaderLen
+	backupPinIDEnd          = backupPinIDStart + backupPinIDBytes
+	backupReadTSStart       = backupPinIDEnd
+	backupReadTSEnd         = backupReadTSStart + backupUint64Size
+	backupDeadlineStart     = backupReadTSEnd
+	backupDeadlineEnd       = backupDeadlineStart + backupUint64Size
+	backupExtendMillisStart = backupPinIDEnd
+	backupExtendMillisEnd   = backupExtendMillisStart + backupUint64Size
+)
+
+var (
+	ErrBackupWireMalformed = errors.New("backup fsm wire payload is malformed")
+	ErrBackupWireSubtype   = errors.New("backup fsm wire subtype is unknown")
+)
+
+type BackupPinEntry struct {
+	PinID    BackupPinID
+	ReadTS   uint64
+	Deadline time.Time
+}
+
+type BackupExtendEntry struct {
+	PinID    BackupPinID
+	Deadline time.Time
+}
+
+type BackupReleaseEntry struct {
+	PinID BackupPinID
+}
+
+type backupEntry struct {
+	subtype byte
+	pin     BackupPinEntry
+	extend  BackupExtendEntry
+	release BackupReleaseEntry
+}
+
+func EncodeBackupPinEntry(entry BackupPinEntry) []byte {
+	out := make([]byte, backupPinEntryLen)
+	out[0] = raftEncodeBackup
+	out[1] = backupSubtypePin
+	copy(out[backupPinIDStart:backupPinIDEnd], entry.PinID[:])
+	binary.BigEndian.PutUint64(out[backupReadTSStart:backupReadTSEnd], entry.ReadTS)
+	binary.BigEndian.PutUint64(out[backupDeadlineStart:backupDeadlineEnd], backupDeadlineMillis(entry.Deadline))
+	return out
+}
+
+func EncodeBackupExtendEntry(entry BackupExtendEntry) []byte {
+	out := make([]byte, backupExtendEntryLen)
+	out[0] = raftEncodeBackup
+	out[1] = backupSubtypeExtend
+	copy(out[backupPinIDStart:backupPinIDEnd], entry.PinID[:])
+	binary.BigEndian.PutUint64(out[backupExtendMillisStart:backupExtendMillisEnd], backupDeadlineMillis(entry.Deadline))
+	return out
+}
+
+func EncodeBackupReleaseEntry(entry BackupReleaseEntry) []byte {
+	out := make([]byte, backupReleaseEntryLen)
+	out[0] = raftEncodeBackup
+	out[1] = backupSubtypeRelease
+	copy(out[backupPinIDStart:backupPinIDEnd], entry.PinID[:])
+	return out
+}
+
+func decodeBackupEntry(data []byte) (backupEntry, error) {
+	if len(data) < backupEnvelopeHeaderLen || data[0] != raftEncodeBackup {
+		return backupEntry{}, errors.WithStack(ErrBackupWireMalformed)
+	}
+	return decodeBackupPayload(data[1:])
+}
+
+func decodeBackupPayload(data []byte) (backupEntry, error) {
+	if len(data) < backupEnvelopeHeaderLen-1 {
+		return backupEntry{}, errors.WithStack(ErrBackupWireMalformed)
+	}
+	switch data[0] {
+	case backupSubtypePin:
+		if len(data) != backupPinEntryLen-1 {
+			return backupEntry{}, errors.WithStack(ErrBackupWireMalformed)
+		}
+		var id BackupPinID
+		copy(id[:], data[backupPinIDStart-1:backupPinIDEnd-1])
+		return backupEntry{
+			subtype: backupSubtypePin,
+			pin: BackupPinEntry{
+				PinID:    id,
+				ReadTS:   binary.BigEndian.Uint64(data[backupReadTSStart-1 : backupReadTSEnd-1]),
+				Deadline: backupDeadlineFromMillis(binary.BigEndian.Uint64(data[backupDeadlineStart-1 : backupDeadlineEnd-1])),
+			},
+		}, nil
+	case backupSubtypeExtend:
+		if len(data) != backupExtendEntryLen-1 {
+			return backupEntry{}, errors.WithStack(ErrBackupWireMalformed)
+		}
+		var id BackupPinID
+		copy(id[:], data[backupPinIDStart-1:backupPinIDEnd-1])
+		return backupEntry{
+			subtype: backupSubtypeExtend,
+			extend: BackupExtendEntry{
+				PinID:    id,
+				Deadline: backupDeadlineFromMillis(binary.BigEndian.Uint64(data[backupExtendMillisStart-1 : backupExtendMillisEnd-1])),
+			},
+		}, nil
+	case backupSubtypeRelease:
+		if len(data) != backupReleaseEntryLen-1 {
+			return backupEntry{}, errors.WithStack(ErrBackupWireMalformed)
+		}
+		var id BackupPinID
+		copy(id[:], data[backupPinIDStart-1:backupPinIDEnd-1])
+		return backupEntry{
+			subtype: backupSubtypeRelease,
+			release: BackupReleaseEntry{
+				PinID: id,
+			},
+		}, nil
+	default:
+		return backupEntry{}, errors.WithStack(ErrBackupWireSubtype)
+	}
+}
+
+func backupDeadlineMillis(deadline time.Time) uint64 {
+	ms := deadline.UnixMilli()
+	if ms <= 0 {
+		return 0
+	}
+	return uint64(ms)
+}
+
+func backupDeadlineFromMillis(ms uint64) time.Time {
+	if ms == 0 {
+		return time.Time{}
+	}
+	if ms > math.MaxInt64 {
+		ms = math.MaxInt64
+	}
+	return time.UnixMilli(int64(ms)) //nolint:gosec // clamped to MaxInt64 above.
+}
diff --git a/kv/backup_codec_test.go b/kv/backup_codec_test.go
new file mode 100644
index 000000000..637e767b0
--- /dev/null
+++ b/kv/backup_codec_test.go
@@ -0,0 +1,91 @@
+package kv
+
+import (
+	"testing"
+	"time"
+
+	"github.com/bootjp/elastickv/internal/encryption/fsmwire"
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestBackupCodecRoundTrip(t *testing.T) {
+	pinID := backupTrackerTestPinID(7)
+	deadline := time.UnixMilli(1780000000123)
+
+	pinWire := EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    pinID,
+		ReadTS:   42,
+		Deadline: deadline,
+	})
+	require.Len(t, pinWire, backupPinEntryLen)
+	gotPin, err := decodeBackupEntry(pinWire)
+	require.NoError(t, err)
+	require.Equal(t, backupSubtypePin, gotPin.subtype)
+	require.Equal(t, BackupPinEntry{PinID: pinID, ReadTS: 42, Deadline: deadline}, gotPin.pin)
+
+	extendWire := EncodeBackupExtendEntry(BackupExtendEntry{
+		PinID:    pinID,
+		Deadline: deadline.Add(time.Second),
+	})
+	require.Len(t, extendWire, backupExtendEntryLen)
+	gotExtend, err := decodeBackupEntry(extendWire)
+	require.NoError(t, err)
+	require.Equal(t, backupSubtypeExtend, gotExtend.subtype)
+	require.Equal(t, BackupExtendEntry{PinID: pinID, Deadline: deadline.Add(time.Second)}, gotExtend.extend)
+
+	releaseWire := EncodeBackupReleaseEntry(BackupReleaseEntry{PinID: pinID})
+	require.Len(t, releaseWire, backupReleaseEntryLen)
+	gotRelease, err := decodeBackupEntry(releaseWire)
+	require.NoError(t, err)
+	require.Equal(t, backupSubtypeRelease, gotRelease.subtype)
+	require.Equal(t, BackupReleaseEntry{PinID: pinID}, gotRelease.release)
+}
+
+func TestBackupCodecRejectsMalformedWire(t *testing.T) {
+	pinID := backupTrackerTestPinID(1)
+	valid := EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    pinID,
+		ReadTS:   42,
+		Deadline: time.UnixMilli(5000),
+	})
+
+	_, err := decodeBackupEntry(nil)
+	require.ErrorIs(t, err, ErrBackupWireMalformed)
+	_, err = decodeBackupEntry([]byte{raftEncodeBatch, backupSubtypeRelease})
+	require.ErrorIs(t, err, ErrBackupWireMalformed)
+	_, err = decodeBackupEntry(valid[:len(valid)-1])
+	require.ErrorIs(t, err, ErrBackupWireMalformed)
+	_, err = decodeBackupEntry([]byte{raftEncodeBackup, 0xff})
+	require.ErrorIs(t, err, ErrBackupWireSubtype)
+}
+
+func TestBackupCodecZeroDeadlineDecodesToZeroTime(t *testing.T) {
+	pinID := backupTrackerTestPinID(1)
+
+	gotPin, err := decodeBackupEntry(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    pinID,
+		ReadTS:   42,
+		Deadline: time.Time{},
+	}))
+	require.NoError(t, err)
+	require.True(t, gotPin.pin.Deadline.IsZero())
+
+	gotExtend, err := decodeBackupEntry(EncodeBackupExtendEntry(BackupExtendEntry{
+		PinID:    pinID,
+		Deadline: time.Time{},
+	}))
+	require.NoError(t, err)
+	require.True(t, gotExtend.extend.Deadline.IsZero())
+}
+
+func TestBackupEnvelopeOpcodeDoesNotCollide(t *testing.T) {
+	require.Equal(t, byte(0x0e), raftEncodeBackup)
+	require.Greater(t, raftEncodeBackup, fsmwire.OpEncryptionMax)
+	require.NotContains(t, []byte{raftEncodeSingle, raftEncodeBatch, raftEncodeHLCLease}, raftEncodeBackup)
+	require.NotContains(t, []byte{0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d}, raftEncodeBackup)
+
+	err := proto.Unmarshal([]byte{raftEncodeBackup, 0x00}, &pb.Request{})
+	require.Error(t, err, "0x0e must remain an invalid proto wire start byte")
+}
diff --git a/kv/fsm.go b/kv/fsm.go
index 421f5fc3b..7a5b3500f 100644
--- a/kv/fsm.go
+++ b/kv/fsm.go
@@ -51,6 +51,11 @@ type kvFSM struct {
 	// write", preserving Stage 6A behavior for backends that did
 	// not opt in).
 	pendingApplyIdx uint64
+	// readTracker is shared with the local FSM compactor. BackupPin
+	// FSM entries mutate this tracker so compaction retains versions
+	// at the live-backup read timestamp until the pin is released or
+	// its deadline expires.
+	readTracker *ActiveTimestampTracker
 	// cutoverSource provides the writer-side view of the Phase-2
 	// envelope cutover index for snapshot v1/v2 selection (Stage
 	// 8a §3.3). nil = always v1 output.
@@ -234,6 +239,12 @@ func WithRouteHistory(routes RouteHistory, shardGroupID uint64) FSMOption {
 	}
 }
 
+func WithActiveTimestampTracker(tracker *ActiveTimestampTracker) FSMOption {
+	return func(f *kvFSM) {
+		f.readTracker = tracker
+	}
+}
+
 // NewKvFSMWithHLC creates a KV FSM that updates hlc.physicalCeiling whenever
 // a HLC lease entry is applied. The caller must pass the same *HLC instance to
 // the coordinator so both sides share the agreed physical ceiling.
@@ -256,6 +267,13 @@ func NewKvFSMWithHLC(store store.MVCCStore, hlc *HLC, opts ...FSMOption) FSM {
 	return f
 }
 
+func NewKvFSMWithHLCAndTracker(store store.MVCCStore, hlc *HLC, tracker *ActiveTimestampTracker, opts ...FSMOption) FSM {
+	all := make([]FSMOption, 0, len(opts)+1)
+	all = append(all, WithActiveTimestampTracker(tracker))
+	all = append(all, opts...)
+	return NewKvFSMWithHLC(store, hlc, all...)
+}
+
 var _ FSM = (*kvFSM)(nil)
 var _ raftengine.StateMachine = (*kvFSM)(nil)
 
@@ -351,6 +369,8 @@ func (f *kvFSM) applyReservedOpcode(data []byte) (any, bool) {
 	switch {
 	case data[0] == raftEncodeHLCLease:
 		return f.applyHLCLease(data[1:]), true
+	case data[0] == raftEncodeBackup:
+		return f.applyBackup(data[1:]), true
 	case data[0] >= fsmwire.OpEncryptionMin && data[0] <= fsmwire.OpEncryptionMax:
 		return f.applyEncryption(f.pendingApplyIdx, data[0], data[1:]), true
 	default:
@@ -630,18 +650,19 @@ func (f *kvFSM) ApplySnapshotHeader(ceiling, cutover uint64) {
 // is monotonic and lives purely in memory. After the cold-start skip
 // gate fires, the engine still delivers WAL committed-tail entries
 // past snapshot.Metadata.Index; without this classifier those
-// volatile entries get dropped along with KV/MVCC duplicates and the
-// post-snapshot ceiling raise is lost. Codex P1 #934 round 7.
+// volatile entries get dropped along with KV/MVCC duplicates. HLC
+// would lose the post-snapshot ceiling raise; backup pins would lose
+// a post-snapshot retention fence.
 //
 // Re-applying KV/MVCC entries would re-execute OCC validation against
 // store state that has already moved past commit_ts, surfacing
-// spurious conflicts. Returning false for any non-HLC payload tag
+// spurious conflicts. Returning false for persistent internal tags
 // preserves that idempotency. Encryption opcodes (0x03..0x07) MUST
 // also return false — they persist DEK state in the encryption
 // sidecar and re-applying would diverge the sidecar's
 // RaftAppliedIndex from the engine's appliedIndex.
 func (f *kvFSM) IsVolatileOnlyPayload(payload []byte) bool {
-	return len(payload) > 0 && payload[0] == raftEncodeHLCLease
+	return len(payload) > 0 && (payload[0] == raftEncodeHLCLease || payload[0] == raftEncodeBackup)
 }
 
 func (f *kvFSM) handleTxnRequest(ctx context.Context, r *pb.Request, commitTS uint64) error {
diff --git a/kv/fsm_backup.go b/kv/fsm_backup.go
new file mode 100644
index 000000000..1d994321e
--- /dev/null
+++ b/kv/fsm_backup.go
@@ -0,0 +1,33 @@
+package kv
+
+import "github.com/cockroachdb/errors"
+
+var ErrBackupApply = errors.New("backup fsm apply failed")
+
+func (f *kvFSM) applyBackup(data []byte) any {
+	if f.readTracker == nil {
+		return haltErr(errors.Wrap(ErrBackupApply, "kv/fsm: backup entry arrived but no ActiveTimestampTracker is wired"))
+	}
+	entry, err := decodeBackupPayload(data)
+	if err != nil {
+		return haltErr(errors.Wrap(errors.Mark(err, ErrBackupApply), "kv/fsm: decode backup entry"))
+	}
+	switch entry.subtype {
+	case backupSubtypePin:
+		err = f.readTracker.ApplyPinWithDeadlineForGroup(entry.pin.PinID, f.shardGroupID, entry.pin.ReadTS, entry.pin.Deadline)
+	case backupSubtypeExtend:
+		err = f.readTracker.ApplyExtendForGroup(entry.extend.PinID, f.shardGroupID, entry.extend.Deadline)
+	case backupSubtypeRelease:
+		f.readTracker.ReleaseBackupPinForGroup(entry.release.PinID, f.shardGroupID)
+		return nil
+	default:
+		err = ErrBackupWireSubtype
+	}
+	if err != nil {
+		if errors.Is(err, ErrInvalidBackupPin) {
+			return err
+		}
+		return haltErr(errors.Wrap(errors.Mark(err, ErrBackupApply), "kv/fsm: apply backup entry"))
+	}
+	return nil
+}
diff --git a/kv/fsm_backup_test.go b/kv/fsm_backup_test.go
new file mode 100644
index 000000000..267f292b3
--- /dev/null
+++ b/kv/fsm_backup_test.go
@@ -0,0 +1,198 @@
+package kv
+
+import (
+	"testing"
+	"time"
+
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestApplyBackupUsesSharedTracker(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	fsm := newBackupTestFSM(t, tracker)
+	pinID := backupTrackerTestPinID(1)
+	now := time.Now()
+	firstDeadline := time.UnixMilli(now.Add(time.Hour).UnixMilli())
+	secondDeadline := time.UnixMilli(now.Add(2 * time.Hour).UnixMilli())
+
+	require.NoError(t, haltApplyOf(fsm.Apply(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    pinID,
+		ReadTS:   42,
+		Deadline: firstDeadline,
+	}))))
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(42), tracker.Oldest())
+
+	require.NoError(t, haltApplyOf(fsm.Apply(EncodeBackupExtendEntry(BackupExtendEntry{
+		PinID:    pinID,
+		Deadline: secondDeadline,
+	}))))
+	gotDeadline, ok := tracker.BackupPinDeadline(pinID)
+	require.True(t, ok)
+	require.Equal(t, secondDeadline, gotDeadline)
+
+	require.NoError(t, haltApplyOf(fsm.Apply(EncodeBackupReleaseEntry(BackupReleaseEntry{PinID: pinID}))))
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(0), tracker.Oldest())
+}
+
+func TestApplyBackupWithoutTrackerHalts(t *testing.T) {
+	fsm, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+	require.True(t, ok)
+
+	err := haltApplyOf(fsm.Apply(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    backupTrackerTestPinID(1),
+		ReadTS:   42,
+		Deadline: time.UnixMilli(5000),
+	})))
+	require.True(t, errors.Is(err, ErrBackupApply), "err = %v", err)
+}
+
+func TestApplyBackupUnknownSubtypeHalts(t *testing.T) {
+	fsm := newBackupTestFSM(t, NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0)))
+
+	err := haltApplyOf(fsm.Apply([]byte{raftEncodeBackup, 0xff}))
+	require.True(t, errors.Is(err, ErrBackupApply), "err = %v", err)
+	require.True(t, errors.Is(err, ErrBackupWireSubtype), "err = %v", err)
+}
+
+func TestApplyBackupInvalidPinReturnsNonFatalError(t *testing.T) {
+	fsm := newBackupTestFSM(t, NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0)))
+
+	resp := fsm.Apply(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    BackupPinID{},
+		ReadTS:   42,
+		Deadline: time.UnixMilli(5000),
+	}))
+	require.NoError(t, haltApplyOf(resp))
+	respErr, ok := resp.(error)
+	require.True(t, ok)
+	require.ErrorIs(t, respErr, ErrInvalidBackupPin)
+}
+
+func TestApplyBackupLimitDoesNotDropCommittedPins(t *testing.T) {
+	tracker := NewActiveTimestampTracker(
+		WithActiveTimestampTrackerSweepInterval(0),
+		WithActiveTimestampTrackerMaxBackupPins(1),
+	)
+	fsm := newBackupTestFSM(t, tracker)
+	require.NoError(t, haltApplyOf(fsm.Apply(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    backupTrackerTestPinID(1),
+		ReadTS:   42,
+		Deadline: time.Now().Add(time.Hour),
+	}))))
+
+	resp := fsm.Apply(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    backupTrackerTestPinID(2),
+		ReadTS:   43,
+		Deadline: time.Now().Add(2 * time.Hour),
+	}))
+
+	require.NoError(t, haltApplyOf(resp))
+	require.Nil(t, resp)
+	require.Equal(t, 2, tracker.ActiveBackupPinCount())
+	require.Equal(t, uint64(42), tracker.Oldest())
+}
+
+func TestApplyBackupMissingExtendIsNoop(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	fsm := newBackupTestFSM(t, tracker)
+
+	resp := fsm.Apply(EncodeBackupExtendEntry(BackupExtendEntry{
+		PinID:    backupTrackerTestPinID(1),
+		Deadline: time.Now().Add(time.Hour),
+	}))
+
+	require.NoError(t, haltApplyOf(resp))
+	require.Nil(t, resp)
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+}
+
+func TestApplyBackupExpiredExtendIsNoop(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	fsm := newBackupTestFSM(t, tracker)
+	pinID := backupTrackerTestPinID(1)
+	require.NoError(t, tracker.PinWithDeadline(pinID, 42, time.Now().Add(-time.Millisecond)))
+
+	resp := fsm.Apply(EncodeBackupExtendEntry(BackupExtendEntry{
+		PinID:    pinID,
+		Deadline: time.Now().Add(time.Hour),
+	}))
+
+	require.NoError(t, haltApplyOf(resp))
+	require.Nil(t, resp)
+	require.Equal(t, 0, tracker.ActiveBackupPinCount())
+}
+
+func TestApplyBackupZeroDeadlineReturnsNonFatalError(t *testing.T) {
+	fsm := newBackupTestFSM(t, NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0)))
+
+	resp := fsm.Apply(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    backupTrackerTestPinID(1),
+		ReadTS:   42,
+		Deadline: time.Time{},
+	}))
+
+	require.NoError(t, haltApplyOf(resp))
+	respErr, ok := resp.(error)
+	require.True(t, ok)
+	require.ErrorIs(t, respErr, ErrInvalidBackupPin)
+}
+
+func TestApplyBackupPinsAreScopedByRaftGroup(t *testing.T) {
+	tracker := NewActiveTimestampTracker(WithActiveTimestampTrackerSweepInterval(0))
+	fsm1 := newBackupTestFSMWithGroup(t, tracker, 1)
+	fsm2 := newBackupTestFSMWithGroup(t, tracker, 2)
+	pinID := backupTrackerTestPinID(1)
+	entry := BackupPinEntry{
+		PinID:    pinID,
+		ReadTS:   42,
+		Deadline: time.Now().Add(time.Hour),
+	}
+
+	require.NoError(t, haltApplyOf(fsm1.Apply(EncodeBackupPinEntry(entry))))
+	require.NoError(t, haltApplyOf(fsm2.Apply(EncodeBackupPinEntry(entry))))
+	require.Equal(t, 2, tracker.ActiveBackupPinCount())
+
+	require.NoError(t, haltApplyOf(fsm1.Apply(EncodeBackupReleaseEntry(BackupReleaseEntry{PinID: pinID}))))
+	require.Equal(t, 1, tracker.ActiveBackupPinCount())
+	_, ok := tracker.BackupPinDeadlineForGroup(pinID, 1)
+	require.False(t, ok)
+	_, ok = tracker.BackupPinDeadlineForGroup(pinID, 2)
+	require.True(t, ok)
+}
+
+func TestBackupPayloadIsVolatileOnly(t *testing.T) {
+	fsm := &kvFSM{}
+	pinID := backupTrackerTestPinID(1)
+
+	require.True(t, fsm.IsVolatileOnlyPayload(EncodeBackupPinEntry(BackupPinEntry{
+		PinID:    pinID,
+		ReadTS:   42,
+		Deadline: time.UnixMilli(5000),
+	})))
+	require.False(t, fsm.IsVolatileOnlyPayload(nil))
+	require.False(t, fsm.IsVolatileOnlyPayload([]byte{raftEncodeSingle}))
+	require.False(t, fsm.IsVolatileOnlyPayload([]byte{0x03}))
+}
+
+func newBackupTestFSM(t *testing.T, tracker *ActiveTimestampTracker) *kvFSM {
+	t.Helper()
+	fsm, ok := NewKvFSMWithHLCAndTracker(store.NewMVCCStore(), NewHLC(), tracker).(*kvFSM)
+	require.True(t, ok)
+	return fsm
+}
+
+func newBackupTestFSMWithGroup(t *testing.T, tracker *ActiveTimestampTracker, groupID uint64) *kvFSM {
+	t.Helper()
+	fsm, ok := NewKvFSMWithHLCAndTracker(
+		store.NewMVCCStore(),
+		NewHLC(),
+		tracker,
+		WithRouteHistory(nil, groupID),
+	).(*kvFSM)
+	require.True(t, ok)
+	return fsm
+}
diff --git a/main.go b/main.go
index 530dbf4b7..f8f8dbd17 100644
--- a/main.go
+++ b/main.go
@@ -386,6 +386,7 @@ func run() error {
 	}
 	keystore := encryption.NewKeystore()
 	redisApplyObserver := adapter.NewRedisApplyObserver()
+	readTracker := kv.NewActiveTimestampTracker()
 
 	// Stage 6D-6c: buildShardGroupsWithEncryptionWiring assembles the
 	// storage-envelope write-path wiring (cipher + deterministic nonce
@@ -412,6 +413,7 @@ func run() error {
 		keystore,
 		*encryptionSidecarPath,
 		*encryptionEnabled,
+		readTracker,
 		cfg.engine,
 		redisApplyObserver,
 	)
@@ -434,9 +436,9 @@ func run() error {
 
 	cleanup := internalutil.CleanupStack{}
 	defer cleanup.Run()
+	cleanup.Add(readTracker.Close)
 
 	ctx, cancel := context.WithCancel(context.Background())
-	readTracker := kv.NewActiveTimestampTracker()
 	shardStore := kv.NewShardStore(cfg.engine, shardGroups)
 	cleanup.Add(func() {
 		_ = shardStore.Close()
@@ -1022,6 +1024,7 @@ func buildShardGroups(
 	factory raftengine.Factory,
 	proposalObserverForGroup func(uint64) kv.ProposalObserver,
 	clock *kv.HLC,
+	readTracker *kv.ActiveTimestampTracker,
 	kekWrapper kek.Wrapper,
 	keystore *encryption.Keystore,
 	sidecarPath string,
@@ -1099,7 +1102,8 @@ func buildShardGroups(
 		// work. At M2 the FSM stores both but does not consult them;
 		// see docs/design/2026_05_29_implemented_composed1_cross_group_commit_guard.md
 		// §M2.
-		sm := kv.NewKvFSMWithHLC(st, clock, fsmOptionsForGroup(applier, routeEngine, g.id, encWiring, applyObservers...)...)
+		sm := kv.NewKvFSMWithHLCAndTracker(st, clock, readTracker,
+			fsmOptionsForGroup(applier, routeEngine, g.id, encWiring, applyObservers...)...)
 		groupBootstrap, groupBootstrapServers, groupBootstrapSeed := bootstrapSettingsForGroup(bootstrapCfg, g.id, bootstrap)
 		runtime, err := buildRuntimeForGroup(
 			raftID, g, raftDir, multi, groupBootstrap,
diff --git a/main_bootstrap_e2e_test.go b/main_bootstrap_e2e_test.go
index d962c8105..1e101daae 100644
--- a/main_bootstrap_e2e_test.go
+++ b/main_bootstrap_e2e_test.go
@@ -540,7 +540,9 @@ func startBootstrapE2ENode(
 		return nil, err
 	}
 	clock := kv.NewHLC()
-	runtimes, shardGroups, err := buildShardGroups(ep.id, baseDir, cfg.groups, cfg.multi, bootstrap, raftBootstrapConfig{legacyServers: bootstrapServers}, factory, nil, clock, nil, nil, "", encryptionWriteWiring{}, cfg.engine)
+	runtimes, shardGroups, err := buildShardGroups(ep.id, baseDir, cfg.groups, cfg.multi, bootstrap,
+		raftBootstrapConfig{legacyServers: bootstrapServers}, factory, nil, clock, kv.NewActiveTimestampTracker(),
+		nil, nil, "", encryptionWriteWiring{}, cfg.engine)
 	if err != nil {
 		return nil, err
 	}
@@ -623,7 +625,9 @@ func startBootstrapE2EMultiGroupNode(
 		return nil, err
 	}
 	clock := kv.NewHLC()
-	runtimes, shardGroups, err := buildShardGroups(ep.id, baseDir, cfg.groups, cfg.multi, bootstrap, bootstrapCfg, factory, nil, clock, nil, nil, "", encryptionWriteWiring{}, cfg.engine)
+	runtimes, shardGroups, err := buildShardGroups(ep.id, baseDir, cfg.groups, cfg.multi, bootstrap,
+		bootstrapCfg, factory, nil, clock, kv.NewActiveTimestampTracker(), nil, nil, "",
+		encryptionWriteWiring{}, cfg.engine)
 	if err != nil {
 		return nil, err
 	}
diff --git a/main_encryption_write_wiring.go b/main_encryption_write_wiring.go
index b3f124e71..165a03b18 100644
--- a/main_encryption_write_wiring.go
+++ b/main_encryption_write_wiring.go
@@ -36,6 +36,7 @@ func buildShardGroupsWithEncryptionWiring(
 	keystore *encryption.Keystore,
 	sidecarPath string,
 	encryptionEnabled bool,
+	readTracker *kv.ActiveTimestampTracker,
 	routeEngine *distribution.Engine,
 	applyObserver kv.ApplyObserver,
 ) ([]*raftGroupRuntime, map[uint64]*kv.ShardGroup, encryptionWriteWiring, error) {
@@ -57,7 +58,8 @@ func buildShardGroupsWithEncryptionWiring(
 	}
 	configureRaftEnvelopeFactory(factory, encWiring)
 	runtimes, shardGroups, err := buildShardGroups(raftID, raftDir, groups, multi, bootstrap, bootstrapCfg,
-		factory, proposalObserverForGroup, clock, kekWrapper, keystore, sidecarPath, encWiring, routeEngine, applyObserver)
+		factory, proposalObserverForGroup, clock, readTracker, kekWrapper, keystore, sidecarPath, encWiring,
+		routeEngine, applyObserver)
 	if err != nil {
 		return runtimes, shardGroups, encWiring, err
 	}
diff --git a/multiraft_runtime_test.go b/multiraft_runtime_test.go
index 7e20233b3..231882c27 100644
--- a/multiraft_runtime_test.go
+++ b/multiraft_runtime_test.go
@@ -54,7 +54,8 @@ func TestBuildShardGroupsWithEtcdEngineRoutesAcrossGroups(t *testing.T) {
 	factory, err := newRaftFactory(raftEngineEtcd, nil)
 	require.NoError(t, err)
 	clock := kv.NewHLC()
-	runtimes, shardGroups, err := buildShardGroups("n1", baseDir, groups, true, true, raftBootstrapConfig{}, factory, nil, clock, nil, nil, "", encryptionWriteWiring{}, nil)
+	runtimes, shardGroups, err := buildShardGroups("n1", baseDir, groups, true, true, raftBootstrapConfig{},
+		factory, nil, clock, kv.NewActiveTimestampTracker(), nil, nil, "", encryptionWriteWiring{}, nil)
 	require.NoError(t, err)
 
 	engine := distribution.NewEngine()
@@ -108,7 +109,9 @@ func TestBuildShardGroupsWithEtcdEngineRestartsAcrossGroups(t *testing.T) {
 	openShardStore := func(bootstrap bool) ([]*raftGroupRuntime, map[uint64]*kv.ShardGroup, *kv.ShardStore) {
 		factory, err := newRaftFactory(raftEngineEtcd, nil)
 		require.NoError(t, err)
-		runtimes, shardGroups, err := buildShardGroups("n1", baseDir, groups, true, bootstrap, raftBootstrapConfig{}, factory, nil, sharedClock, nil, nil, "", encryptionWriteWiring{}, nil)
+		runtimes, shardGroups, err := buildShardGroups("n1", baseDir, groups, true, bootstrap,
+			raftBootstrapConfig{}, factory, nil, sharedClock, kv.NewActiveTimestampTracker(), nil, nil, "",
+			encryptionWriteWiring{}, nil)
 		require.NoError(t, err)
 		shardStore := kv.NewShardStore(engine, shardGroups)
 		return runtimes, shardGroups, shardStore