Fix deferred BUFFER_ONE overlap processing (#11555)

To editorialize, the presence and difficulty of spotting these bugs
makes me want to refactor this code more, but I'm going to defer that
for now. Anyway, checked the V1 code's equivalent and also checked that
the integration-test catches the problem before applying fix.

## LLM Summary

- include an existing deferred `BUFFER_ONE` start in overlap resolution
- keep the earliest buffered occurrence and reject later arrivals
- retain catchup-window expiry for an already-deferred occurrence

## Problem

`BUFFER_ONE` permits at most one pending occurrence while an action
workflow is running. CHASM represents buffered-start lifecycle with
`Attempt`:

- `0`: newly enqueued and not processed;
- `-1`: processed but deferred while another workflow is running;
- `1+`: executing or retrying.

Before this change, `InvokerProcessBufferTask` passed only `Attempt ==
0` starts to overlap resolution. This produced the following sequence:

1. A workflow is running.
2. Occurrence A becomes due under `BUFFER_ONE`.
3. A is retained and marked deferred with `Attempt == -1`.
4. Occurrence B becomes due before the workflow closes.
5. Processing ignores A and presents only B to the shared overlap
resolver.
6. The resolver sees no occupied one-element buffer and retains B as
well.

The invoker can therefore contain both A and B, violating `BUFFER_ONE`
and potentially executing an unexpected workflow.

## Fix

Include a deferred start in the pending overlap set when its effective
overlap policy resolves to `BUFFER_ONE`:

```go
return start.Attempt == 0 ||
    (start.Attempt == -1 &&
        scheduler.resolveOverlapPolicy(start.GetOverlapPolicy()) ==
            enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
```

This lets the existing shared V1 `ProcessBuffer` logic see the occupied
buffer, retain the earliest occurrence, and reject later arrivals. The
effective policy is resolved because a buffered start may store
`UNSPECIFIED` and inherit the schedule current policy.

The special handling is limited to `BUFFER_ONE`; other deferred starts
should not be broadly reprocessed merely because a new occurrence
arrived.

## How this was found

The Schedule V1-to-CHASM production-history replay initially reported
5,355 V1 actions versus 4,961 CHASM actions, suggesting CHASM had
skipped 394 workflows.

Tracing the first independently different action showed the opposite:
CHASM emitted one extra occurrence. Because that workflow did not exist
in V1 history, the replay had no completion event to apply to it. It
remained running in simulated CHASM state and blocked hundreds of later
starts. The apparent 394-action deficit was a downstream alternate-chain
cascade, not 394 independent defects.

Inspecting state at the first difference revealed one running workflow,
one deferred `BUFFER_ONE` start (`Attempt == -1`), one new start
(`Attempt == 0`), and only the new start participating in overlap
resolution. A focused unit test reproduced that exact state.

After the fix, the representative converged to 5,355 actions on both
sides with identical workflow identities; only observation-time
differences remained.

## Impact

The direct product impact is one additional retained and potentially
executed workflow. This is high severity because `BUFFER_ONE` explicitly
bounds pending work, and an unexpected workflow may perform externally
visible or non-idempotent actions.

## Testing

- `TestProcessBufferTask_BufferOneKeepsExistingDeferredStart` verifies
that the first deferred occurrence occupies the buffer and the later
occurrence is rejected.
- `TestProcessBufferTask_BufferOneDropsDeferredStartPastCatchupWindow`
verifies that a deferred occurrence is still dropped when its catchup
deadline has expired.

```sh
go test -tags test_dep ./chasm/lib/scheduler \
  -run "^TestProcessBufferTask_BufferOne" -count=1
```

chasm/lib/scheduler/invoker_process_buffer_task_test.go

@@ -14,6 +14,7 @@ import (
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/service/history/tasks"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -317,6 +318,58 @@ func TestProcessBufferTask_BufferOne(t *testing.T) {
 	})
 }
 
+func TestProcessBufferTask_BufferOneKeepsExistingDeferredStart(t *testing.T) {
+	env := newTestEnv(t)
+	startTime := timestamppb.New(env.TimeSource.Now())
+	runProcessBufferTestCase(t, env, &processBufferTestCase{
+		InitialBufferedStarts: []*schedulespb.BufferedStart{
+			{
+				NominalTime:   startTime,
+				ActualTime:    startTime,
+				DesiredTime:   startTime,
+				RequestId:     "deferred-first",
+				WorkflowId:    "deferred-first",
+				OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
+				Attempt:       -1,
+			},
+			{
+				NominalTime:   startTime,
+				ActualTime:    startTime,
+				DesiredTime:   startTime,
+				RequestId:     "new-later",
+				WorkflowId:    "new-later",
+				OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
+			},
+		},
+		InitialRunningWorkflows:  []*commonpb.WorkflowExecution{{WorkflowId: "running", RunId: "running-run"}},
+		ExpectedBufferedStarts:   1,
+		ExpectedRunningWorkflows: 1,
+		ExpectedOverlapSkipped:   1,
+		ValidateInvoker: func(t *testing.T, invoker *scheduler.Invoker) {
+			require.Equal(t, "deferred-first", invoker.GetBufferedStarts()[0].GetRequestId())
+			require.Equal(t, int64(-1), invoker.GetBufferedStarts()[0].GetAttempt())
+		},
+	})
+}
+
+func TestProcessBufferTask_BufferOneDropsDeferredStartPastCatchupWindow(t *testing.T) {
+	env := newTestEnv(t)
+	env.Scheduler.Schedule.Policies.CatchupWindow = durationpb.New(10 * time.Minute)
+	startTime := timestamppb.New(env.TimeSource.Now().Add(-15 * time.Minute))
+	runProcessBufferTestCase(t, env, &processBufferTestCase{
+		InitialBufferedStarts: []*schedulespb.BufferedStart{{
+			NominalTime:   startTime,
+			ActualTime:    startTime,
+			DesiredTime:   startTime,
+			RequestId:     "deferred-expired",
+			WorkflowId:    "deferred-expired",
+			OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
+			Attempt:       -1,
+		}},
+		ExpectedMissedCatchupWindow: 1,
+	})
+}
+
 // ProcessBuffer is scheduled with an empty buffer.
 func TestProcessBufferTask_Empty(t *testing.T) {
 	env := newTestEnv(t)

chasm/lib/scheduler/invoker_tasks.go

@@ -501,9 +501,11 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer(
 	isRunning := len(runningWorkflows) > 0
 	result.missedCatchupByActionRunning = make(map[bool]int64)
 
-	// Processing completely ignores any BufferedStart that's already executing/backing off.
+	// Processing ignores starts that are already executing or backing off. An existing
+	// deferred BUFFER_ONE start still participates so it can reject later starts.
 	pendingBufferedStarts := util.FilterSlice(invoker.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool {
-		return start.Attempt == 0
+		return start.Attempt == 0 ||
+			(start.Attempt == -1 && scheduler.resolveOverlapPolicy(start.GetOverlapPolicy()) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
 	})
 
 	// Resolve overlap policies and trim BufferedStarts that are skipped by policy.

tests/schedule_test.go

@@ -769,10 +769,9 @@ func testFutureActionTimesAdvanceWhilePaused(t *testing.T, newContext contextFac
 }
 
 // testBufferOneDeferredFiresAfterCompletion exercises the BUFFER_ONE deferred
-// lifecycle end-to-end: an action that gets buffered while a workflow is
-// running must fire once that workflow completes. Without re-enabling the
-// deferred start (Attempt=-1 -> 0 in recordCompletedAction), the buffered
-// fire would be stranded.
+// lifecycle end-to-end. Later ticks must not displace or accumulate alongside
+// the first buffered action, and that action must fire once the running workflow
+// completes.
 func testBufferOneDeferredFiresAfterCompletion(t *testing.T, newContext contextFactory) {
 	s := newScheduleEnv(t, scheduleCommonOpts(t)...)
 
@@ -804,6 +803,19 @@ func testBufferOneDeferredFiresAfterCompletion(t *testing.T, newContext contextF
 	}, awaitTimeout, pollInterval, "expected exactly one running workflow with one deferred start buffered behind it")
 	require.Equal(t, int32(1), runs.Load(), "only the first workflow should have fired before the running one completes")
 
+	// Keep the first workflow open across several more ticks. V1 evaluates the
+	// complete buffer and keeps its first entry; CHASM must do the same even after
+	// that entry has been marked deferred (Attempt=-1).
+	require.Never(t, func() bool {
+		desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
+			Namespace:  s.Namespace().String(),
+			ScheduleId: sid,
+		})
+		return descErr == nil && (desc.GetInfo().GetBufferSize() != 1 ||
+			len(desc.GetInfo().GetRunningWorkflows()) != 1 || runs.Load() != 1)
+	}, 3*fastInterval, pollInterval,
+		"V1 and CHASM must retain exactly one buffered occurrence while later ticks arrive")
+
 	// Releasing the running workflow must re-enable the deferred start (Attempt=-1 -> 0) so it fires.
 	require.Equal(t, 1, completeRunningWorkflows(ctx, t, s, sid))
 	await.RequireTruef(t, func() bool { return runs.Load() == 2 },