From b61aa17914a218636fbf6dc6ee13a47f0e360865 Mon Sep 17 00:00:00 2001
From: dawn <dawn@tangled.org>
Date: Fri, 28 Aug 2026 22:14:57 +0900
Subject: [PATCH] spindle/executor: stop sending events after a job finishes

Signed-off-by: dawn <dawn@tangled.org>
---
 spindle/mill/executor/executor.go      |  1 +
 spindle/mill/executor/outbox.go        | 21 ++++++++-
 spindle/mill/executor/reserved_test.go | 65 ++++++++++++++++++++++++++
 3 files changed, 86 insertions(+), 1 deletion(-)

diff --git a/spindle/mill/executor/executor.go b/spindle/mill/executor/executor.go
index 419f3c03..051d0818 100644
--- a/spindle/mill/executor/executor.go
+++ b/spindle/mill/executor/executor.go
@@ -58,6 +58,7 @@ type Executor struct {
 	outboxBytes    int64
 	maxOutboxBytes int64 // 10 MiB default outbox cap
 	outboxIdleCh   chan struct{}
+	terminalSeqnos map[string]uint64
 
 	eventMu sync.Mutex
 	sendMu  sync.Mutex
diff --git a/spindle/mill/executor/outbox.go b/spindle/mill/executor/outbox.go
index 3de4396f..9938a625 100644
--- a/spindle/mill/executor/outbox.go
+++ b/spindle/mill/executor/outbox.go
@@ -51,8 +51,13 @@ func (e *Executor) initOutbox() error {
 		return fmt.Errorf("list outbox rows: %w", err)
 	}
 
+	e.terminalSeqnos = make(map[string]uint64)
 	for _, r := range rows {
 		e.outboxBytes += r.ByteSize
+		entry, err := decodeEvent(r.Payload, r.Seqno)
+		if err == nil && entry.GetAttemptResult() != nil {
+			e.terminalSeqnos[entry.GetLeaseId()] = r.Seqno
+		}
 	}
 	if e.outboxBytes > 0 {
 		e.outboxIdleCh = make(chan struct{})
@@ -91,15 +96,24 @@ func (e *Executor) appendAndSend(leaseID string, payload any, control bool) erro
 	}
 
 	e.eventMu.Lock()
+	if _, terminal := e.terminalSeqnos[leaseID]; terminal {
+		e.l.Warn("dropping stream entry after terminal", "lease", leaseID)
+		e.eventMu.Unlock()
+		return nil
+	}
 	if control && !isTerminal && e.maxOutboxBytes > 0 && e.outboxBytes+int64(len(encoded)) > e.maxOutboxBytes {
 		e.l.Warn("outbox reserve exhausted; dropping nonterminal status", "cap", e.maxOutboxBytes)
 		e.eventMu.Unlock()
 		return nil
 	}
-	if _, err := e.db.AppendOutboxRow(encoded, control); err != nil {
+	seqno, err := e.db.AppendOutboxRow(encoded, control)
+	if err != nil {
 		e.eventMu.Unlock()
 		return fmt.Errorf("append outbox row: %w", err)
 	}
+	if isTerminal {
+		e.terminalSeqnos[leaseID] = seqno
+	}
 	if e.outboxBytes == 0 {
 		e.outboxIdleCh = make(chan struct{})
 	}
@@ -352,6 +366,11 @@ func (e *Executor) deleteOutboxPrefix(upTo uint64) error {
 	deleted, err := e.db.DeleteOutboxPrefix(upTo)
 	if err == nil {
 		e.subtractOutboxBytes(deleted)
+		for leaseID, seqno := range e.terminalSeqnos {
+			if seqno <= upTo {
+				delete(e.terminalSeqnos, leaseID)
+			}
+		}
 		if e.outboxBytes == 0 && e.outboxIdleCh != nil {
 			close(e.outboxIdleCh)
 			e.outboxIdleCh = nil
diff --git a/spindle/mill/executor/reserved_test.go b/spindle/mill/executor/reserved_test.go
index 582e6ac6..7ed1c8de 100644
--- a/spindle/mill/executor/reserved_test.go
+++ b/spindle/mill/executor/reserved_test.go
@@ -288,6 +288,68 @@ func TestHandleCancelFinalizesExpiredReservation(t *testing.T) {
 	}
 }
 
+func TestTerminalOutboxEntrySuppressesLaterEvents(t *testing.T) {
+	e := testExecutor(t)
+	if err := e.appendTerminal("lease-1", string(models.StatusKindSuccess), nil); err != nil {
+		t.Fatal(err)
+	}
+
+	e.handleCancel("lease-1")
+	if err := e.appendStatus("lease-1", &tangled.PipelineStatus{Status: string(models.StatusKindRunning)}); err != nil {
+		t.Fatal(err)
+	}
+
+	rows, err := e.db.ListOutboxRows()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("outbox rows after terminal = %d, want 1", len(rows))
+	}
+	var entry millv1.Event
+	if err := proto.Unmarshal(rows[0].Payload, &entry); err != nil {
+		t.Fatal(err)
+	}
+	if got := entry.GetAttemptResult().GetStatus(); got != millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS {
+		t.Fatalf("terminal status = %v, want SUCCESS", got)
+	}
+}
+
+func TestTerminalGuardSurvivesExecutorRestart(t *testing.T) {
+	d := testDB(t)
+	first := &Executor{
+		db:             d,
+		l:              slog.New(slog.NewTextHandler(io.Discard, nil)),
+		active:         make(map[string]*reservation),
+		maxOutboxBytes: 10 * 1024 * 1024,
+	}
+	if err := first.initOutbox(); err != nil {
+		t.Fatal(err)
+	}
+	if err := first.appendTerminal("lease-1", string(models.StatusKindSuccess), nil); err != nil {
+		t.Fatal(err)
+	}
+
+	restarted := &Executor{
+		db:             d,
+		l:              slog.New(slog.NewTextHandler(io.Discard, nil)),
+		active:         make(map[string]*reservation),
+		maxOutboxBytes: 10 * 1024 * 1024,
+	}
+	if err := restarted.initOutbox(); err != nil {
+		t.Fatal(err)
+	}
+	restarted.handleCancel("lease-1")
+
+	rows, err := d.ListOutboxRows()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("outbox rows after restart and duplicate cancel = %d, want 1", len(rows))
+	}
+}
+
 func TestHandleCommitPreservesPreauthorizedSecrets(t *testing.T) {
 	d := testDB(t)
 	n := notifier.New()
@@ -590,6 +652,9 @@ func TestReplayDropsRowsAlreadyAcknowledgedByResume(t *testing.T) {
 	if len(rows) != 0 {
 		t.Fatalf("outbox rows after resume = %d, want 0", len(rows))
 	}
+	if len(e.terminalSeqnos) != 0 {
+		t.Fatalf("terminal guards after resume = %d, want 0", len(e.terminalSeqnos))
+	}
 	select {
 	case <-e.outboxIdle():
 	default:
-- 
2.54.0