From d62e6b7fe832b99dd77d960a8decc826d6eef1ab Mon Sep 17 00:00:00 2001
From: dawn <dawn@tangled.org>
Date: Wed, 2 Sep 2026 22:31:51 +0900
Subject: [PATCH] spindle/db,spindle/mill: retry transient sqlite lock errors

Signed-off-by: dawn <dawn@tangled.org>
---
 spindle/db/mill_state.go      | 217 +++++++++++++++++++---------------
 spindle/db/mill_state_test.go |  50 ++++++++
 spindle/db/retry.go           |  30 +++++
 spindle/db/retry_test.go      |  75 ++++++++++++
 spindle/mill/mill.go          |  82 +++++++------
 5 files changed, 324 insertions(+), 130 deletions(-)
 create mode 100644 spindle/db/retry.go
 create mode 100644 spindle/db/retry_test.go

diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go
index 7009a5376..c11becda5 100644
--- a/spindle/db/mill_state.go
+++ b/spindle/db/mill_state.go
@@ -111,96 +111,105 @@ func (d *DB) ListExecutorCursors() ([]ExecutorCursor, error) {
 }
 
 func (d *DB) SetOutboxEpoch(epoch string) error {
-	tx, err := d.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
+	return retrySQLite(func() error {
+		tx, err := d.Begin()
+		if err != nil {
+			return err
+		}
+		defer tx.Rollback()
 
-	if _, err := tx.Exec(`delete from mill_outbox_rows`); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(`delete from mill_outbox_state`); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(`insert into mill_outbox_state (epoch, next_seqno) values (?, 1)`, epoch); err != nil {
-		return err
-	}
-	return tx.Commit()
+		if _, err := tx.Exec(`delete from mill_outbox_rows`); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(`delete from mill_outbox_state`); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(`insert into mill_outbox_state (epoch, next_seqno) values (?, 1)`, epoch); err != nil {
+			return err
+		}
+		return tx.Commit()
+	})
 }
 
 func (d *DB) AppendOutboxRow(payload []byte, control bool) (uint64, error) {
-	tx, err := d.Begin()
-	if err != nil {
-		return 0, err
-	}
-	defer tx.Rollback()
-
-	var epoch string
 	var nextSeqno uint64
-	err = tx.QueryRow(`
-		update mill_outbox_state
-		set next_seqno = next_seqno + 1
-		where epoch = (select epoch from mill_outbox_state limit 1)
-		returning epoch, next_seqno - 1
-	`).Scan(&epoch, &nextSeqno)
-	if err == sql.ErrNoRows {
-		return 0, fmt.Errorf("no outbox epoch set")
-	} else if err != nil {
-		return 0, err
-	}
+	err := retrySQLite(func() error {
+		tx, err := d.Begin()
+		if err != nil {
+			return err
+		}
+		defer tx.Rollback()
 
-	byteSize := int64(len(payload))
-	controlVal := 0
-	if control {
-		controlVal = 1
-	}
+		var epoch string
+		err = tx.QueryRow(`
+			update mill_outbox_state
+			set next_seqno = next_seqno + 1
+			where epoch = (select epoch from mill_outbox_state limit 1)
+			returning epoch, next_seqno - 1
+		`).Scan(&epoch, &nextSeqno)
+		if err == sql.ErrNoRows {
+			return fmt.Errorf("no outbox epoch set")
+		} else if err != nil {
+			return err
+		}
 
-	if _, err := tx.Exec(
-		`insert into mill_outbox_rows (epoch, seqno, payload, byte_size, control) values (?, ?, ?, ?, ?)`,
-		epoch, nextSeqno, payload, byteSize, controlVal,
-	); err != nil {
-		return 0, err
-	}
+		byteSize := int64(len(payload))
+		controlVal := 0
+		if control {
+			controlVal = 1
+		}
 
-	if err := tx.Commit(); err != nil {
+		if _, err := tx.Exec(
+			`insert into mill_outbox_rows (epoch, seqno, payload, byte_size, control) values (?, ?, ?, ?, ?)`,
+			epoch, nextSeqno, payload, byteSize, controlVal,
+		); err != nil {
+			return err
+		}
+
+		return tx.Commit()
+	})
+	if err != nil {
 		return 0, err
 	}
 	return nextSeqno, nil
 }
 
 func (d *DB) DeleteOutboxPrefix(ackedSeqno uint64) (OutboxDeletion, error) {
-	tx, err := d.Begin()
-	if err != nil {
-		return OutboxDeletion{}, err
-	}
-	defer tx.Rollback()
+	var deleted OutboxDeletion
+	err := retrySQLite(func() error {
+		tx, err := d.Begin()
+		if err != nil {
+			return err
+		}
+		defer tx.Rollback()
 
-	var epoch string
-	err = tx.QueryRow(`select epoch from mill_outbox_state limit 1`).Scan(&epoch)
-	if err == sql.ErrNoRows {
-		return OutboxDeletion{}, nil
-	}
-	if err != nil {
-		return OutboxDeletion{}, err
-	}
+		var epoch string
+		err = tx.QueryRow(`select epoch from mill_outbox_state limit 1`).Scan(&epoch)
+		if err == sql.ErrNoRows {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
 
-	var deleted OutboxDeletion
-	if err := tx.QueryRow(`
-		select count(*),
-		       coalesce(sum(byte_size), 0)
-		from mill_outbox_rows
-		where epoch = ? and seqno <= ?
-	`, epoch, ackedSeqno).Scan(&deleted.Rows, &deleted.Bytes); err != nil {
-		return OutboxDeletion{}, err
-	}
-	if _, err := tx.Exec(
-		`delete from mill_outbox_rows where epoch = ? and seqno <= ?`,
-		epoch, ackedSeqno,
-	); err != nil {
-		return OutboxDeletion{}, err
-	}
-	if err := tx.Commit(); err != nil {
+		deleted = OutboxDeletion{}
+		if err := tx.QueryRow(`
+			select count(*),
+			       coalesce(sum(byte_size), 0)
+			from mill_outbox_rows
+			where epoch = ? and seqno <= ?
+		`, epoch, ackedSeqno).Scan(&deleted.Rows, &deleted.Bytes); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(
+			`delete from mill_outbox_rows where epoch = ? and seqno <= ?`,
+			epoch, ackedSeqno,
+		); err != nil {
+			return err
+		}
+		return tx.Commit()
+	})
+	if err != nil {
 		return OutboxDeletion{}, err
 	}
 	return deleted, nil
@@ -418,31 +427,53 @@ func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) {
 	return res, rows.Err()
 }
 
-func (d *DB) ApplyEventBatch(n *notifier.Notifier, fn func(tx *EventBatchTx) error) error {
-	err := eventstream.WithClock(func(insert eventstream.Inserter) error {
-		tx, err := d.Begin()
-		if err != nil {
-			return err
-		}
-		defer tx.Rollback()
-
-		batchTx := &EventBatchTx{
-			tx:     tx,
-			db:     d,
-			insert: insert,
-		}
-		if err := fn(batchTx); err != nil {
-			return err
-		}
-		return tx.Commit()
+// ApplyEventBatch runs fn in a transaction and retries it while sqlite reports
+// transient lock errors. fn may run more than once, so non-transactional side
+// effects must be returned through its result: the result is published only
+// from the attempt that commits, and a failed attempt's result is discarded.
+func ApplyEventBatch[T any](d *DB, n *notifier.Notifier, fn func(tx *EventBatchTx) (T, error)) (T, error) {
+	var result T
+	err := retrySQLite(func() error {
+		return eventstream.WithClock(func(insert eventstream.Inserter) error {
+			tx, err := d.Begin()
+			if err != nil {
+				return err
+			}
+			defer tx.Rollback()
+
+			batchTx := &EventBatchTx{
+				tx:     tx,
+				db:     d,
+				insert: insert,
+			}
+			v, err := fn(batchTx)
+			if err != nil {
+				return err
+			}
+			if err := tx.Commit(); err != nil {
+				return err
+			}
+			result = v
+			return nil
+		})
 	})
 	if err != nil {
-		return err
+		var zero T
+		return zero, err
 	}
 	if n != nil {
 		n.NotifyAll()
 	}
-	return nil
+	return result, nil
+}
+
+// ApplyEventBatch is the package-level ApplyEventBatch for callers without a
+// result.
+func (d *DB) ApplyEventBatch(n *notifier.Notifier, fn func(tx *EventBatchTx) error) error {
+	_, err := ApplyEventBatch(d, n, func(tx *EventBatchTx) (struct{}, error) {
+		return struct{}{}, fn(tx)
+	})
+	return err
 }
 
 func (d *DB) DeleteMillLeasesByRepo(repoDid string) ([]string, error) {
diff --git a/spindle/db/mill_state_test.go b/spindle/db/mill_state_test.go
index 39adb1ddd..7ac1b7f66 100644
--- a/spindle/db/mill_state_test.go
+++ b/spindle/db/mill_state_test.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/mattn/go-sqlite3"
 	"tangled.org/core/notifier"
 	"tangled.org/core/spindle/models"
 )
@@ -88,6 +89,55 @@ func TestExecutorCursors(t *testing.T) {
 
 }
 
+func TestApplyEventBatchPublishesResultOnlyOnCommit(t *testing.T) {
+	d := newTestDB(t)
+	attempts := 0
+	res, err := ApplyEventBatch(d, nil, func(tx *EventBatchTx) ([]string, error) {
+		attempts++
+		if attempts == 1 {
+			// a result produced by a failed attempt must never escape
+			return []string{"ghost"}, sqlite3.Error{Code: sqlite3.ErrBusy}
+		}
+		if err := tx.AdvanceCursor("node-1", "inc-1", 7); err != nil {
+			return nil, err
+		}
+		return []string{"committed"}, nil
+	})
+	if err != nil {
+		t.Fatalf("ApplyEventBatch: %v", err)
+	}
+	if attempts != 2 {
+		t.Fatalf("callback attempts = %d, want 2", attempts)
+	}
+	if len(res) != 1 || res[0] != "committed" {
+		t.Fatalf("result = %v, want [committed]", res)
+	}
+	if got, err := d.GetExecutorCursor("node-1", "inc-1"); err != nil || got != 7 {
+		t.Fatalf("cursor = %d, err = %v; want 7", got, err)
+	}
+}
+
+func TestApplyEventBatchRetriesBusy(t *testing.T) {
+	d := newTestDB(t)
+	attempts := 0
+	err := d.ApplyEventBatch(nil, func(tx *EventBatchTx) error {
+		attempts++
+		if attempts == 1 {
+			return sqlite3.Error{Code: sqlite3.ErrBusy}
+		}
+		return tx.AdvanceCursor("node-1", "inc-1", 5)
+	})
+	if err != nil {
+		t.Fatalf("ApplyEventBatch: %v", err)
+	}
+	if attempts != 2 {
+		t.Fatalf("callback attempts = %d, want 2", attempts)
+	}
+	if got, err := d.GetExecutorCursor("node-1", "inc-1"); err != nil || got != 5 {
+		t.Fatalf("cursor = %d, err = %v; want 5", got, err)
+	}
+}
+
 func TestCompleteMillLeaseIsAtomic(t *testing.T) {
 	d := newTestDB(t)
 	lease := MillLease{
diff --git a/spindle/db/retry.go b/spindle/db/retry.go
new file mode 100644
index 000000000..7146c4705
--- /dev/null
+++ b/spindle/db/retry.go
@@ -0,0 +1,30 @@
+package db
+
+import (
+	"errors"
+	"time"
+
+	"github.com/mattn/go-sqlite3"
+)
+
+const sqliteRetryAttempts = 4
+
+func isSQLiteBusy(err error) bool {
+	var sqliteErr sqlite3.Error
+	if !errors.As(err, &sqliteErr) {
+		return false
+	}
+	return sqliteErr.Code == sqlite3.ErrBusy || sqliteErr.Code == sqlite3.ErrLocked
+}
+
+func retrySQLite(fn func() error) error {
+	var err error
+	for attempt := 0; attempt < sqliteRetryAttempts; attempt++ {
+		err = fn()
+		if !isSQLiteBusy(err) || attempt == sqliteRetryAttempts-1 {
+			return err
+		}
+		time.Sleep(time.Duration(10*(1<<attempt)) * time.Millisecond)
+	}
+	return err
+}
diff --git a/spindle/db/retry_test.go b/spindle/db/retry_test.go
new file mode 100644
index 000000000..72362ae5b
--- /dev/null
+++ b/spindle/db/retry_test.go
@@ -0,0 +1,75 @@
+package db
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// regression for the 2026-09-02 executor-lost incident: a writer blocked past
+// the busy timeout by a competing connection must retry and apply exactly
+// once, not surface "database is locked" and kill the executor session
+func TestAppendOutboxRowRetriesHeldWriteLock(t *testing.T) {
+	ctx := context.Background()
+	dbPath := filepath.Join(t.TempDir(), "mill.db")
+	d, err := Make(ctx, dbPath)
+	if err != nil {
+		t.Fatalf("Make: %v", err)
+	}
+	t.Cleanup(func() { d.Close() })
+
+	if err := d.SetOutboxEpoch("epoch-1"); err != nil {
+		t.Fatalf("SetOutboxEpoch: %v", err)
+	}
+
+	other, err := Make(ctx, dbPath)
+	if err != nil {
+		t.Fatalf("Make second: %v", err)
+	}
+	t.Cleanup(func() { other.Close() })
+	conn, err := other.Conn(ctx)
+	if err != nil {
+		t.Fatalf("Conn: %v", err)
+	}
+	defer conn.Close()
+	if _, err := conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil {
+		t.Fatalf("BEGIN IMMEDIATE: %v", err)
+	}
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		// hold the write lock past the 5s busy timeout so the first attempt
+		// fails and retrySQLite must recover
+		time.Sleep(5500 * time.Millisecond)
+		_, _ = conn.ExecContext(ctx, "ROLLBACK")
+	}()
+
+	start := time.Now()
+	seqno, err := d.AppendOutboxRow([]byte("payload"), false)
+	if err != nil {
+		t.Fatalf("AppendOutboxRow: %v", err)
+	}
+	if elapsed := time.Since(start); elapsed < 5*time.Second {
+		t.Fatalf("AppendOutboxRow returned in %v without contending", elapsed)
+	}
+	if seqno != 1 {
+		t.Fatalf("seqno = %d, want 1", seqno)
+	}
+
+	epoch, next, err := d.GetOutboxState()
+	if err != nil {
+		t.Fatalf("GetOutboxState: %v", err)
+	}
+	if epoch != "epoch-1" || next != 2 {
+		t.Fatalf("outbox state = (%q, %d), want (epoch-1, 2): seqno must advance exactly once across retries", epoch, next)
+	}
+	rows, err := d.ListOutboxRows()
+	if err != nil {
+		t.Fatalf("ListOutboxRows: %v", err)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("outbox rows = %d, want 1", len(rows))
+	}
+	<-done
+}
diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go
index ea5d36b3b..2b09f8171 100644
--- a/spindle/mill/mill.go
+++ b/spindle/mill/mill.go
@@ -1314,15 +1314,10 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 		lease *RemoteLease
 		ar    *millv1.AttemptResult
 	}
-	var pendingTerminals []pendingTerminal
-	var terminalCancelIDs []string
-	var artifactLeases []*RemoteLease
-	finishedInBatch := make(map[string]struct{})
 	type startupObservation struct {
 		engine string
 		delay  time.Duration
 	}
-	var startupObservations []startupObservation
 	leaseSet := make(map[*RemoteLease]struct{})
 	for _, entry := range newEntries {
 		m.mu.Lock()
@@ -1352,12 +1347,22 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 		}
 	}
 
-	var highestSeqno uint64 = current
+	type batchApply struct {
+		terminalCancelIDs   []string
+		pendingTerminals    []pendingTerminal
+		artifactLeases      []*RemoteLease
+		startupObservations []startupObservation
+		highestSeqno        uint64
+	}
+
+	applyFunc := func(tx *db.EventBatchTx) (batchApply, error) {
+		var r batchApply
+		r.highestSeqno = current
+		finishedInBatch := make(map[string]struct{})
 
-	applyFunc := func(tx *db.EventBatchTx) error {
 		for _, entry := range newEntries {
 			if entry.GetAttemptResult() != nil {
-				terminalCancelIDs = append(terminalCancelIDs, entry.LeaseId)
+				r.terminalCancelIDs = append(r.terminalCancelIDs, entry.LeaseId)
 			}
 			m.mu.Lock()
 			lease := m.leases[entry.LeaseId]
@@ -1366,13 +1371,13 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 			// events for leases this node doesn't own are skipped but still
 			// count as processed
 			if lease == nil || lease.nodeID != sess.nodeID {
-				highestSeqno = entry.Seqno
+				r.highestSeqno = entry.Seqno
 				continue
 			}
 
 			// events arriving for a different epoch are invalid (different session)
 			if lease.epoch != "" && lease.epoch != sess.epoch {
-				return protoErrf("lease %q epoch %q does not match session %q", lease.id, lease.epoch, sess.epoch)
+				return batchApply{}, protoErrf("lease %q epoch %q does not match session %q", lease.id, lease.epoch, sess.epoch)
 			}
 
 			lease.mu.Lock()
@@ -1381,12 +1386,12 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 
 			// done leases can replay terminals on reconnect, skip them
 			if state == leaseDone {
-				highestSeqno = entry.Seqno
+				r.highestSeqno = entry.Seqno
 				continue
 			}
 
 			if _, finished := finishedInBatch[lease.id]; finished {
-				return protoErrf("stream entry follows terminal for lease %q", lease.id)
+				return batchApply{}, protoErrf("stream entry follows terminal for lease %q", lease.id)
 			}
 
 			switch {
@@ -1405,18 +1410,18 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 				if tx != nil {
 					alreadyRunning, err := tx.HasWorkflowStatus(context.Background(), pipelineAtUri, lease.wid.Name, statusStr)
 					if err != nil {
-						return err
+						return batchApply{}, err
 					}
 					if err := tx.InsertStatusEvent(pipelineAtUri, lease.wid.Name, statusStr, errMsg, exitCode); err != nil {
-						return err
+						return batchApply{}, err
 					}
 					if !alreadyRunning {
 						delay, ok, err := tx.WorkflowStartupDelay(context.Background(), pipelineAtUri, lease.wid.Name)
 						if err != nil {
-							return err
+							return batchApply{}, err
 						}
 						if ok {
-							startupObservations = append(startupObservations, startupObservation{
+							r.startupObservations = append(r.startupObservations, startupObservation{
 								engine: lease.engine,
 								delay:  delay,
 							})
@@ -1437,7 +1442,7 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 				case millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED:
 					statusStr = "cancelled"
 				default:
-					return protoErrf("unsupported terminal status %v", ar.Status)
+					return batchApply{}, protoErrf("unsupported terminal status %v", ar.Status)
 				}
 				var errMsg *string
 				if e := ar.GetError(); e != "" {
@@ -1451,56 +1456,59 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 				pipelineAtUri := string(lease.wid.PipelineId.AtUri())
 				if tx != nil {
 					if err := tx.InsertStatusEvent(pipelineAtUri, lease.wid.Name, statusStr, errMsg, exitCode); err != nil {
-						return err
+						return batchApply{}, err
 					}
 					if err := tx.DeleteLease(lease.id); err != nil {
-						return err
+						return batchApply{}, err
 					}
 					if a := ar.GetLogArtifact(); a != nil {
 						if a.GetRef() == "" {
-							return protoErrf("empty log artifact ref")
+							return batchApply{}, protoErrf("empty log artifact ref")
 						}
 						// the ref names an object the mill deletes on wipe;
 						// pin it to the derived log key so an executor cannot
 						// point cleanup at someone else's objects
 						if a.GetRef() != "logs/"+lease.id+".log" {
-							return protoErrf("log artifact ref %q does not match lease %s", a.GetRef(), lease.id)
+							return batchApply{}, protoErrf("log artifact ref %q does not match lease %s", a.GetRef(), lease.id)
 						}
 						if !strings.HasPrefix(a.GetHash(), "sha256:") {
-							return protoErrf("invalid log artifact hash %q", a.GetHash())
+							return batchApply{}, protoErrf("invalid log artifact hash %q", a.GetHash())
 						}
 						if lease.wid.Knot == "" || lease.wid.Rkey == "" || lease.wid.Name == "" {
 							m.l.Error("skipping log artifact with incomplete workflow identity", "lease", lease.id, "wid", lease.wid)
 						} else {
 							if tx != nil {
 								if err := tx.InsertArtifactRef(lease.id, lease.repoDID, lease.wid, a.GetRef(), a.GetHash()); err != nil {
-									return err
+									return batchApply{}, err
 								}
 							}
-							artifactLeases = append(artifactLeases, lease)
+							r.artifactLeases = append(r.artifactLeases, lease)
 						}
 					}
 				}
-				pendingTerminals = append(pendingTerminals, pendingTerminal{
+				r.pendingTerminals = append(r.pendingTerminals, pendingTerminal{
 					lease: lease,
 					ar:    ar,
 				})
 			}
 
-			highestSeqno = entry.Seqno
+			r.highestSeqno = entry.Seqno
 		}
 
 		if tx != nil {
-			return tx.AdvanceCursor(sess.nodeID, sess.epoch, highestSeqno)
+			if err := tx.AdvanceCursor(sess.nodeID, sess.epoch, r.highestSeqno); err != nil {
+				return batchApply{}, err
+			}
 		}
-		return nil
+		return r, nil
 	}
 
+	var applied batchApply
 	var err error
 	if m.db != nil {
-		err = m.db.ApplyEventBatch(m.n, applyFunc)
+		applied, err = db.ApplyEventBatch(m.db, m.n, applyFunc)
 	} else {
-		err = applyFunc(nil)
+		applied, err = applyFunc(nil)
 	}
 
 	if err != nil {
@@ -1508,12 +1516,12 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 		return err
 	}
 
-	for _, observation := range startupObservations {
+	for _, observation := range applied.startupObservations {
 		m.metrics.RecordWorkflowStartupDelay(context.Background(), observation.engine, "mill", observation.delay)
 	}
 
 	if m.cfg.LogDir != "" {
-		for _, lease := range artifactLeases {
+		for _, lease := range applied.artifactLeases {
 			path := models.LogFilePath(m.cfg.LogDir, lease.wid)
 			if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
 				m.l.Warn("failed to remove live log file after artifact recorded", "path", path, "err", err)
@@ -1522,15 +1530,15 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 	}
 
 	m.mu.Lock()
-	if highestSeqno > m.nodeSeqno[currentKey] {
-		m.nodeSeqno[currentKey] = highestSeqno
+	if applied.highestSeqno > m.nodeSeqno[currentKey] {
+		m.nodeSeqno[currentKey] = applied.highestSeqno
 	}
 	m.mu.Unlock()
 
-	for _, leaseID := range terminalCancelIDs {
+	for _, leaseID := range applied.terminalCancelIDs {
 		m.settleSessionCancel(sess, leaseID)
 	}
-	for _, pt := range pendingTerminals {
+	for _, pt := range applied.pendingTerminals {
 		if pt.lease.millRecordsTerminalMetrics &&
 			pt.ar.GetMillRecordsTerminalMetrics() &&
 			pt.ar.GetFailureClass() != "" &&
@@ -1555,7 +1563,7 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error {
 	}
 	unlockLeases()
 
-	return m.sendAck(sess, highestSeqno)
+	return m.sendAck(sess, applied.highestSeqno)
 }
 func terminalMetricResult(status millv1.TerminalStatus) string {
 	switch status {
-- 
2.51.2