fix(runtime): keep a stalled queue delivery from running after a newer turn

debounce_test.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -530,3 +531,62 @@ func TestDebounceSupersededWaiterExitsBeforeItsInterval(t *testing.T) {
 		t.Fatalf("handler ran before any quiet period elapsed, calls = %d", calls)
 	}
 }
+
+type validationGatedAdapter struct {
+	*fakeAdapter
+	calls   atomic.Int32
+	reached chan struct{}
+	release chan struct{}
+}
+
+func (a *validationGatedAdapter) ValidateThreadID(id chat.ThreadID) (chat.ThreadRef, error) {
+	if a.calls.Add(1) == 1 {
+		close(a.reached)
+		<-a.release
+	}
+	return a.fakeAdapter.ValidateThreadID(id)
+}
+
+func TestDebounceDeliveryDelayedInValidationCannotDisplaceNewerWaiter(t *testing.T) {
+	t.Parallel()
+
+	adapter := &validationGatedAdapter{
+		fakeAdapter: newFakeAdapter("fake"),
+		reached:     make(chan struct{}),
+		release:     make(chan struct{}),
+	}
+	var logs syncBuffer
+	bot := newDebounceRuntime(t, newFakeState(), adapter, &logs, func(o *chat.RuntimeOptions) {
+		o.DebounceInterval = 500 * time.Millisecond
+	})
+	var mu sync.Mutex
+	var handled []string
+	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
+		mu.Lock()
+		handled = append(handled, ev.Event.ID)
+		mu.Unlock()
+		return nil
+	})
+
+	olderDone := make(chan postEventResult, 1)
+	go func() { olderDone <- postEventResultFor(bot, "fake", mentionEvent("older", "fake:v1:thread-1")) }()
+	waitClosed(t, adapter.reached, "older delivery did not reach thread ID validation")
+	if status := postEvent(t, bot, "fake", mentionEvent("newer", "fake:v1:thread-1")); status != http.StatusOK {
+		t.Fatalf("newer status = %d", status)
+	}
+	close(adapter.release)
+	if res := <-olderDone; res.err != nil || res.status != http.StatusOK {
+		t.Fatalf("older result status=%d err=%v", res.status, res.err)
+	}
+	eventually(t, 3*time.Second, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(handled) >= 1
+	}, "debounce winner did not dispatch")
+
+	mu.Lock()
+	defer mu.Unlock()
+	if !equalStrings(handled, []string{"newer"}) {
+		t.Fatalf("handled = %v, want only [newer]: a delivery delayed in validation displaced the final event", handled)
+	}
+}

queue_edge_test.go

@@ -733,3 +733,124 @@ func TestQueueOlderLockTakerKeepsNewerWaiter(t *testing.T) {
 		})
 	}
 }
+
+type gatedMarkState struct {
+	*gatedAcquireState
+	gateID      string
+	markReached chan struct{}
+	markRelease chan struct{}
+}
+
+func newGatedMarkState(gateID string) *gatedMarkState {
+	return &gatedMarkState{
+		gatedAcquireState: newGatedAcquireState(0),
+		gateID:            gateID,
+		markReached:       make(chan struct{}),
+		markRelease:       make(chan struct{}),
+	}
+}
+
+func (s *gatedMarkState) MarkEvent(ctx context.Context, id string, ttl time.Duration) (bool, error) {
+	if id == s.gateID {
+		close(s.markReached)
+		<-s.markRelease
+	}
+	return s.gatedAcquireState.MarkEvent(ctx, id, ttl)
+}
+
+func TestQueueDeliveryStalledBeforeRegisteringNeverRunsAfterNewerLockTaker(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range queueDispatchModes {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			state := newGatedMarkState("X")
+			run := newGatedQueueRun(t, state, tc.mode, "A")
+
+			run.post("A")
+			waitClosed(t, run.blockStarted, "A's handler did not start")
+			run.post("X")
+			waitClosed(t, state.markReached, "X did not reach its dedupe mark after the Lock Conflict")
+			close(run.releaseBlock)
+			eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) },
+				"A did not release the Thread Lock")
+			run.post("C")
+			eventually(t, 2*time.Second, func() bool {
+				got := run.handledIDs()
+				return len(got) == 2 && got[1] == "C" && !state.lockHeld(string(queueThread))
+			}, "C did not take the free lock and run")
+			close(state.markRelease)
+			eventually(t, 4*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
+				"X did not reach a terminal outcome")
+
+			if got := run.handledIDs(); !equalStrings(got, []string{"A", "C"}) {
+				t.Fatalf("handled = %v, want [A C]: X was admitted before C, so C's turn supersedes it", got)
+			}
+		})
+	}
+}
+
+func TestQueueDeliveryStalledInItsLockCallNeverRunsAfterNewerLockTaker(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range queueDispatchModes {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			const xPreludeAcquire = 1
+			state := newGatedAcquireState(xPreludeAcquire)
+			run := newGatedQueueRun(t, state, tc.mode, "")
+
+			run.post("X")
+			waitClosed(t, state.reached, "X's prelude did not reach the lock")
+			run.post("C")
+			eventually(t, 2*time.Second, func() bool {
+				return equalStrings(run.handledIDs(), []string{"C"}) && !state.lockHeld(string(queueThread))
+			}, "C did not take the free lock and run")
+			close(state.release)
+			eventually(t, 2*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 2 },
+				"X did not reach a terminal outcome")
+
+			if got := run.handledIDs(); !equalStrings(got, []string{"C"}) {
+				t.Fatalf("handled = %v, want [C]: X was admitted before C, so C's turn supersedes it", got)
+			}
+			if state.lockHeld(string(queueThread)) {
+				t.Fatal("X kept the Thread Lock after it ended as superseded")
+			}
+		})
+	}
+}
+
+func TestQueueDeliveryStalledBeforeRegisteringNeverRunsAfterNewerWaiter(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range queueDispatchModes {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			const cFirstPoll = 4 // after the prelude acquires of A, X, and C
+			state := newGatedMarkState("X")
+			run := newGatedQueueRun(t, state, tc.mode, "A")
+
+			run.post("A")
+			waitClosed(t, run.blockStarted, "A's handler did not start")
+			run.post("X")
+			waitClosed(t, state.markReached, "X did not reach its dedupe mark after the Lock Conflict")
+			run.post("C")
+			eventually(t, 2*time.Second, func() bool { return state.acquireCalls() >= cFirstPoll },
+				"C did not park and poll for the lock")
+			close(run.releaseBlock)
+			eventually(t, 2*time.Second, func() bool {
+				return equalStrings(run.handledIDs(), []string{"A", "C"}) && !state.lockHeld(string(queueThread))
+			}, "C did not take its turn after A")
+			close(state.markRelease)
+			eventually(t, 4*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
+				"X did not reach a terminal outcome")
+
+			if got := run.handledIDs(); !equalStrings(got, []string{"A", "C"}) {
+				t.Fatalf("handled = %v, want [A C]: X was admitted before C, so C's turn supersedes it", got)
+			}
+		})
+	}
+}