[Scheduled Actions] Nexus-based workflow completions (watcher) (#8424)

- Adds the Nexus-based workflow watcher to Scheduler V2.
- Scheduler V2 now maintains a mapping of request IDs -> workflow IDs,
recorded within the same transaction as the buffered start, that can be
used to identify workflow completion results.
- Workflow ID generation is moved from within `startWorkflow` to within
`SpecProcessor` (earlier, but identical algorithm/results), so that it
can be recorded at the same time as the BufferedStart.
- `LastCompletionState` is added as a CHASM field (separate document) to
keep track of payload results.
- A Nexus completion callback is attached to workflows started via
Invoker.

---------

Co-authored-by: Roey Berman <[email protected]>

Author: lina-temporal

api/schedule/v1/message.pb.go

@@ -149,6 +149,7 @@ func (b *BackfillerTaskExecutor) processBackfill(
 		startTime,
 		endTime,
 		request.GetOverlapPolicy(),
+		scheduler.WorkflowID(),
 		backfiller.GetBackfillId(),
 		true,
 		&limit,

chasm/lib/scheduler/backfiller_tasks.go

@@ -12,7 +12,6 @@ import (
 	"go.temporal.io/server/chasm/lib/scheduler"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	"go.temporal.io/server/common/metrics"
-	"go.temporal.io/server/service/history/tasks"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -241,7 +240,6 @@ func (s *backfillerTasksSuite) runTestCase(c *backfillTestCase) {
 	// Either type of request will spawn a Backfiller and schedule an immediate pure task.
 	_, err = s.node.CloseTransaction()
 	s.NoError(err)
-	s.True(s.hasTask(&tasks.ChasmTaskPure{}, chasm.TaskScheduledTimeImmediate))
 
 	// Run a backfill task.
 	err = s.executor.Execute(ctx, backfiller, chasm.TaskAttributes{}, &schedulerpb.BackfillerTask{})
@@ -268,6 +266,11 @@ func (s *backfillerTasksSuite) runTestCase(c *backfillTestCase) {
 	// Validate BufferedStarts. More detailed validation must be done in the callbacks.
 	s.Equal(c.ExpectedBufferedStarts, len(invoker.GetBufferedStarts()))
 
+	// Validate RequestId -> WorkflowId mapping
+	for _, start := range invoker.GetBufferedStarts() {
+		s.Equal(start.WorkflowId, invoker.WorkflowID(start.RequestId))
+	}
+
 	// Callbacks.
 	if c.ValidateInvoker != nil {
 		c.ValidateInvoker(invoker)

chasm/lib/scheduler/backfiller_tasks_test.go

@@ -0,0 +1,19 @@
+package scheduler
+
+import (
+	"time"
+
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/chasm"
+)
+
+// Export unexported methods for testing.
+
+func (s *Scheduler) RecordCompletedAction(
+	ctx chasm.MutableContext,
+	scheduleTime time.Time,
+	workflowID string,
+	workflowStatus enumspb.WorkflowExecutionStatus,
+) {
+	s.recordCompletedAction(ctx, scheduleTime, workflowID, workflowStatus)
+}