diff --git a/debounce_test.go b/debounce_test.go index 7ff16b6..215c3b0 100644 --- a/debounce_test.go +++ b/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) + } +} diff --git a/queue_edge_test.go b/queue_edge_test.go index bdee078..84422b1 100644 --- a/queue_edge_test.go +++ b/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) + } + }) + } +} diff --git a/runtime.go b/runtime.go index fba2665..2ab7b13 100644 --- a/runtime.go +++ b/runtime.go @@ -20,6 +20,13 @@ const ( // ConcurrencyQueue waits for the in-flight handler, then dispatches the most // recent superseded event for the scope. A newer event that finds the lock // free runs at once and supersedes the waiting event. + // + // "Most recent" follows the dispatch admission order on this instance. A + // delivery delayed in its prelude (lock, routing, dedupe) never runs after a + // newer delivery that already took the scope's turn: it ends as superseded. + // The work before the lock scope is known (the Observer callback and thread + // ID validation) must not block: a delivery delayed there until every newer + // delivery for the scope has finished still runs afterwards. ConcurrencyQueue // ConcurrencyDebounce coalesces rapid follow-ups: each new routed event for a // scope supersedes the previous waiter and only the final event in a @@ -233,9 +240,11 @@ type Chat struct { // inflight tracks detached tails so Shutdown drains them before state shutdown. inflight sync.WaitGroup - // queueMu guards pending, the per-scope most-recent pending waiter. - queueMu sync.Mutex - pending map[string]*pendingWaiter + // queueMu guards pending, the per-scope most-recent pending waiter, and + // queueTurns, the per-scope turn record of the queue strategy. + queueMu sync.Mutex + pending map[string]*pendingWaiter + queueTurns map[string]*queueTurn // dispatchSeq orders deliveries by dispatch admission so a delivery delayed // in its prelude can never displace a newer pending waiter. dispatchSeq atomic.Uint64 @@ -306,6 +315,7 @@ func New(ctx context.Context, opts ...Option) (*Chat, error) { options: cfg.options, eventAcceptances: map[string]*eventAcceptance{}, pending: map[string]*pendingWaiter{}, + queueTurns: map[string]*queueTurn{}, burstScopes: map[string]*burstScope{}, } if cfg.options.Concurrency == ConcurrencyConcurrent { @@ -635,13 +645,21 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error { c.safeEnd(span, OutcomeAdmissionRejected) return fmt.Errorf("chat: dispatch admission rejected for adapter %q: %w", event.Adapter, ErrAdmissionRejected) } + // A queue delivery leaves its scope's turn record once it reaches its + // outcome: when the deferred tail frees its admission slot, or when + // dispatchSync returns. + entry := &queueEntry{} if c.options.Dispatch == DispatchDeferred { - return c.dispatchDeferred(ctx, event, seq, release) + return c.dispatchDeferred(ctx, event, seq, entry, func() { + c.leaveQueueScope(entry) + release() + }) } // A sync delivery retains no work for Shutdown to drain: its caller (an // HTTP server or a Listener) owns its lifetime. release() - return c.dispatchSync(ctx, event, seq) + defer c.leaveQueueScope(entry) + return c.dispatchSync(ctx, event, seq, entry) } // dispatchSync runs the prelude and the routed handler inline under the request @@ -649,8 +667,8 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error { // strategy a Lock Conflict waits inline (bounded by ctx) for the in-flight // handler before the routed handler runs. Under the concurrent strategy the // request waits inline for a MaxConcurrent slot instead of a lock. -func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64) error { - work, resolved, err := c.prelude(ctx, event, seq) +func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64, entry *queueEntry) error { + work, resolved, err := c.prelude(ctx, event, seq, entry) if err != nil || resolved { return err } @@ -686,8 +704,8 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64) error // dispatchDeferred runs the prelude under the request context and, on a routed // event, launches the detached tail and returns so the adapter can acknowledge // the platform (ack-then-work). release frees the delivery's admission slot. -func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64, release func()) error { - work, resolved, err := c.prelude(ctx, event, seq) +func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64, entry *queueEntry, release func()) error { + work, resolved, err := c.prelude(ctx, event, seq, entry) if err != nil || resolved { // The delivery retains nothing past the prelude: an errored or // resolved (duplicate, dropped, ignored, unrouted) prelude frees its @@ -768,7 +786,7 @@ type preludeWork struct { // Under the queue strategy a Lock Conflict on a routed event does not block: the // event is registered as pending (no lease) and returned with needsLock=true so // the tail acquires the lock, keeping ack prompt under DispatchDeferred. -func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64) (preludeWork, bool, error) { +func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *queueEntry) (preludeWork, bool, error) { spanCtx, span := c.safeDispatch(ctx, AdapterAttr(event.Adapter), TenantAttr(event.Tenant)) ctx = spanCtx @@ -812,6 +830,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64) (preludeWo return preludeWork{}, true, err } scope := c.lockScopeKey(event, ref) + c.enterQueueScope(scope, entry) // The prelude acquires the Thread Lock only under drop/queue; debounce and // burst always coordinate in the tail (keeping ack prompt and free of lock @@ -979,7 +998,8 @@ func (c *Chat) routedWork( } // resolveStale closes a routed event whose delivery was admitted before the - // scope's current pending waiter: it never displaces the newer waiter. + // scope's current pending waiter or, under queue, before the scope's latest + // turn: it never displaces the newer waiter or runs after the newer turn. resolveStale := func(label string) (preludeWork, bool, error) { c.logger.Debug("chat "+label+" waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) c.safeEnd(span, OutcomeIgnored, RouteAttr(route)) @@ -1002,7 +1022,10 @@ func (c *Chat) routedWork( work.displaced = displaced return work, false, nil } - c.supersedeOlderWaiter(scope, event, seq) + if !c.takeQueueTurn(scope, event, seq) { + releaseOnResolve() + return resolveStale("queue") + } return work, false, nil case ConcurrencyDebounce: @@ -1297,19 +1320,75 @@ type pendingWaiter struct { displaced chan struct{} } +// queueTurn is a lock scope's turn record under the queue strategy. won is the +// admission sequence of the newest delivery that took a turn (it holds the lock +// and will run its handler). active counts the scope's deliveries from +// enterQueueScope to their outcome. The record lives while active > 0, so a +// counted delivery admitted before won ends stale. +type queueTurn struct { + active int + won uint64 +} + +// queueEntry is the lock scope a queue delivery entered, or "" if it entered +// none: dispatch leaves that scope once the delivery reaches its outcome. +type queueEntry struct { + scope string +} + +// enterQueueScope counts a queue delivery in its scope's turn record once its +// lock scope is known, before any State call. Only non-blocking work (the +// Observer callback and thread ID validation) runs between admission and here. +func (c *Chat) enterQueueScope(scope string, entry *queueEntry) { + if c.options.Concurrency != ConcurrencyQueue { + return + } + assert(entry.scope == "", "a delivery enters one lock scope") + c.queueMu.Lock() + defer c.queueMu.Unlock() + turn := c.queueTurns[scope] + if turn == nil { + turn = &queueTurn{} + c.queueTurns[scope] = turn + } + turn.active++ + entry.scope = scope +} + +// leaveQueueScope ends a queue delivery's count in its scope's turn record and +// drops the record when no delivery for the scope is left. +func (c *Chat) leaveQueueScope(entry *queueEntry) { + if entry.scope == "" { + return + } + c.queueMu.Lock() + defer c.queueMu.Unlock() + turn := c.queueTurns[entry.scope] + assert(turn != nil && turn.active > 0, "a queue delivery leaves a scope it entered") + turn.active-- + if turn.active == 0 { + assert(c.pending[entry.scope] == nil, "a scope with no delivery has no pending waiter") + delete(c.queueTurns, entry.scope) + } +} + // registerPending records event as the most-recent pending event for its scope, // superseding (and signalling) any earlier waiter. "Most recent" follows the // dispatch admission sequence, not registration time: a delivery delayed in // its prelude (validation, dedupe, routing) never displaces a waiter admitted // after it, and instead reports registered=false so the caller resolves it as -// superseded. Recording supersession here, where the displacing event's id is -// known, lets the record carry superseded_by. label names the coordination -// mode ("queue" or "debounce") so the observation names the strategy that -// displaced the waiter. The returned channel closes when this registration is -// itself superseded. +// superseded. Under queue it also reports registered=false when the scope's +// latest turn was admitted after it. Recording supersession here, where the +// displacing event's id is known, lets the record carry superseded_by. label +// names the coordination mode ("queue" or "debounce") so the observation names +// the strategy that displaced the waiter. The returned channel closes when +// this registration is itself superseded. func (c *Chat) registerPending(scope string, event *Event, seq uint64, label string) (displaced <-chan struct{}, registered bool) { c.queueMu.Lock() defer c.queueMu.Unlock() + if turn := c.queueTurns[scope]; turn != nil && turn.won > seq { + return nil, false + } if prev := c.pending[scope]; prev != nil { if prev.seq > seq { return nil, false @@ -1327,16 +1406,27 @@ func (c *Chat) displaceLocked(prev *pendingWaiter, by *Event, label string) { close(prev.displaced) } -func (c *Chat) supersedeOlderWaiter(scope string, event *Event, seq uint64) { +// takeQueueTurn gives event, which took the free lock in its prelude, its +// scope's turn: it supersedes an older pending waiter and makes every delivery +// admitted before event stale. It reports false, and takes no turn, when a +// newer delivery already took one. +func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) bool { c.queueMu.Lock() defer c.queueMu.Unlock() + turn := c.queueTurns[scope] + assert(turn != nil, "a queue delivery takes a turn in a scope it entered") + if turn.won > seq { + return false + } + turn.won = seq prev := c.pending[scope] if prev == nil || prev.seq > seq { - return + return true } assert(prev.seq != seq, "a delivery that holds its prelude lease is never its scope's pending waiter") c.displaceLocked(prev, event, "queue") delete(c.pending, scope) + return true } // isPending reports whether event is still its scope's most-recent pending @@ -1365,12 +1455,11 @@ func (c *Chat) queueForLock(ctx context.Context, scope string, event *Event, lab // AcquireLock was in flight must win, or a superseded debounce/queue // waiter would dispatch alongside the final one. Once the holder passes // this check it is committed; later arrivals wait on the lock. - if !c.isPending(scope, event) { + if !c.commitPending(scope, event) { c.releaseLock(ctx, lease, event.ThreadID) c.logger.Debug("chat "+label+" waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) return LockLease{}, acquireSuperseded } - c.clearPending(scope, event) case acquireSuperseded: // registerPending already emitted the superseded_by record; this waiter // just stops. @@ -1415,6 +1504,25 @@ func (c *Chat) pollForLock(ctx context.Context, scope string, superseded func() } } +// commitPending commits event, which holds the lock, if it is still its scope's +// pending waiter: it clears the entry and, under queue, gives event the scope's +// turn. The check and the clear are one queueMu section, so no newer event can +// displace a waiter that goes on to run. +func (c *Chat) commitPending(scope string, event *Event) bool { + c.queueMu.Lock() + defer c.queueMu.Unlock() + waiter := c.pending[scope] + if waiter == nil || waiter.event != event { + return false + } + delete(c.pending, scope) + if turn := c.queueTurns[scope]; turn != nil { + assert(turn.won < waiter.seq, "a pending waiter is newer than every turn its scope took") + turn.won = waiter.seq + } + return true +} + func (c *Chat) clearPending(scope string, event *Event) { c.queueMu.Lock() defer c.queueMu.Unlock()