test(dispatcher): guard tenant-stream msg IDs against drift (#4547)

* test(dispatcher): guard tenant-stream msg IDs against drift

The pub/sub split (#4480) made the implicit durable-send mirror to tenant
streams an explicit allowlist (msgqueue.tenantStreamMsgIDs), which nearly
dropped task-cancelled on the API cancellation path during review. Nothing
kept that allowlist in sync with what the dispatcher's streams consume.

Make the dispatcher the checkable source of truth: msgsToWorkflowEvent's
switch becomes a lookup in a package-level workflowEventConverters map and
isMatchingWorkflowRunV1's switch a lookup in workflowRunMatchers, preserving
behavior. msgqueue exports the allowlist via TenantStreamMsgIDs(), and
TestTenantStreamMsgIDsInSync asserts set equality between the two, with
failure messages saying exactly what to update on either side.

* refactor(dispatcher): extract tenant-stream event handling into its own file

Move workflowEventConverters, workflowRunMatchers, msgsToWorkflowEvent and
isMatchingWorkflowRunV1 out of server.go into tenant_stream_events.go, next
to the drift test that asserts their keys against msgqueue's allowlist.

Collapse the repeated decode/loop/append scaffolding into two small generic
adapters (eventConverter, runMatcher), so each map entry is just its payload
type and field mapping; the workflow-run-finished status switch stays inside
its closure. Drop the unused DispatcherImpl receivers from both functions.

No behavior change: map misses and empty match results return exactly what
they did before.

Author: igor-kupczynski

internal/msgqueue/pubsub.go

@@ -3,6 +3,8 @@ package msgqueue
 import (
 	"context"
 	"fmt"
+	"maps"
+	"slices"
 
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
@@ -80,9 +82,12 @@ type PubSub interface {
 }
 
 // tenantStreamMsgIDs enumerates the message IDs the dispatcher's gRPC streams
-// consume from tenant topics (see msgsToWorkflowEvent and
-// isMatchingWorkflowRunV1 in the dispatcher). Publishing any other ID to a
+// consume from tenant topics (see workflowEventConverters and
+// workflowRunMatchers in the dispatcher). Publishing any other ID to a
 // tenant topic is pure waste: the streams are the only consumers.
+//
+// Guarded against drift by TestTenantStreamMsgIDsInSync in the dispatcher
+// package: an ID consumed there but missing here is silently never published.
 var tenantStreamMsgIDs = map[string]struct{}{
 	MsgIDCreatedTask:                  {},
 	MsgIDTaskCompleted:                {},
@@ -93,6 +98,13 @@ var tenantStreamMsgIDs = map[string]struct{}{
 	MsgIDWorkflowRunFinishedCandidate: {},
 }
 
+// TenantStreamMsgIDs returns the message IDs PubTenantMessage publishes to
+// tenant-stream topics, sorted. It exists so the dispatcher can assert its
+// stream consumers stay in sync with this allowlist.
+func TenantStreamMsgIDs() []string {
+	return slices.Sorted(maps.Keys(tenantStreamMsgIDs))
+}
+
 // PubTenantMessage writes a tenant-scoped message to its destinations: a
 // durable send to queue via mq (when queue is non-nil), plus a publish to the
 // tenant stream when the message ID is one the dispatcher's streams consume.

internal/services/dispatcher/server.go

@@ -1124,7 +1124,7 @@ func (s *DispatcherImpl) subscribeToWorkflowRunsV1(server contracts.Dispatcher_S
 		wg.Add(1)
 		defer wg.Done()
 
-		if matchedWorkflowRunIds, ok := s.isMatchingWorkflowRunV1(msg, acks); ok {
+		if matchedWorkflowRunIds, ok := isMatchingWorkflowRunV1(msg, acks); ok {
 			if err := iter(matchedWorkflowRunIds); err != nil {
 				s.l.Error().Ctx(ctx).Err(err).Msg("could not iterate over workflow runs")
 			}
@@ -2049,7 +2049,7 @@ func (s *DispatcherImpl) subscribeToWorkflowEventsByWorkflowRunIdV1(workflowRunI
 		wg.Add(1)
 		defer wg.Done()
 
-		events, err := s.msgsToWorkflowEvent(
+		events, err := msgsToWorkflowEvent(
 			msgId,
 			payloads,
 			func(events []*contracts.WorkflowEvent) ([]*contracts.WorkflowEvent, error) {
@@ -2175,7 +2175,7 @@ func (s *DispatcherImpl) subscribeToWorkflowEventsByAdditionalMetaV1(key string,
 		wg.Add(1)
 		defer wg.Done()
 
-		events, err := s.msgsToWorkflowEvent(
+		events, err := msgsToWorkflowEvent(
 			msgId,
 			payloads,
 			func(events []*contracts.WorkflowEvent) ([]*contracts.WorkflowEvent, error) {
@@ -2349,164 +2349,3 @@ func (s *DispatcherImpl) listWorkflowRuns(ctx context.Context, tenantId uuid.UUI
 
 	return res, nil
 }
-
-func (s *DispatcherImpl) msgsToWorkflowEvent(msgId string, payloads [][]byte, filter func(tasks []*contracts.WorkflowEvent) ([]*contracts.WorkflowEvent, error), hangupFunc func(tasks []*contracts.WorkflowEvent) ([]*contracts.WorkflowEvent, error)) ([]*contracts.WorkflowEvent, error) {
-	workflowEvents := []*contracts.WorkflowEvent{}
-
-	switch msgId {
-	case "created-task":
-		payloads := msgqueue.JSONConvert[tasktypes.CreatedTaskPayload](payloads)
-
-		for _, payload := range payloads {
-			workflowEvents = append(workflowEvents, &contracts.WorkflowEvent{
-				WorkflowRunId:  payload.WorkflowRunID.String(),
-				ResourceType:   contracts.ResourceType_RESOURCE_TYPE_STEP_RUN,
-				ResourceId:     payload.ExternalID.String(),
-				EventType:      contracts.ResourceEventType_RESOURCE_EVENT_TYPE_STARTED,
-				EventTimestamp: timestamppb.New(payload.InsertedAt.Time),
-				RetryCount:     &payload.RetryCount,
-			})
-		}
-	case "task-completed":
-		payloads := msgqueue.JSONConvert[tasktypes.CompletedTaskPayload](payloads)
-
-		for _, payload := range payloads {
-			workflowEvents = append(workflowEvents, &contracts.WorkflowEvent{
-				WorkflowRunId:  payload.WorkflowRunId.String(),
-				ResourceType:   contracts.ResourceType_RESOURCE_TYPE_STEP_RUN,
-				ResourceId:     payload.ExternalId.String(),
-				EventType:      contracts.ResourceEventType_RESOURCE_EVENT_TYPE_COMPLETED,
-				EventTimestamp: timestamppb.New(time.Now()),
-				RetryCount:     &payload.RetryCount,
-				EventPayload:   string(payload.Output),
-			})
-		}
-	case "task-failed":
-		payloads := msgqueue.JSONConvert[tasktypes.FailedTaskPayload](payloads)
-
-		for _, payload := range payloads {
-			workflowEvents = append(workflowEvents, &contracts.WorkflowEvent{
-				WorkflowRunId:  payload.WorkflowRunId.String(),
-				ResourceType:   contracts.ResourceType_RESOURCE_TYPE_STEP_RUN,
-				ResourceId:     payload.ExternalId.String(),
-				EventType:      contracts.ResourceEventType_RESOURCE_EVENT_TYPE_FAILED,
-				EventTimestamp: timestamppb.New(time.Now()),
-				RetryCount:     &payload.RetryCount,
-				EventPayload:   payload.ErrorMsg,
-			})
-		}
-	case "task-cancelled":
-		payloads := msgqueue.JSONConvert[tasktypes.CancelledTaskPayload](payloads)
-
-		for _, payload := range payloads {
-			workflowEvents = append(workflowEvents, &contracts.WorkflowEvent{
-				WorkflowRunId:  payload.WorkflowRunId.String(),
-				ResourceType:   contracts.ResourceType_RESOURCE_TYPE_STEP_RUN,
-				ResourceId:     payload.ExternalId.String(),
-				EventType:      contracts.ResourceEventType_RESOURCE_EVENT_TYPE_CANCELLED,
-				EventTimestamp: timestamppb.New(time.Now()),
-				RetryCount:     &payload.RetryCount,
-			})
-		}
-	case "task-stream-event":
-		payloads := msgqueue.JSONConvert[tasktypes.StreamEventPayload](payloads)
-
-		for _, payload := range payloads {
-			workflowEvents = append(workflowEvents, &contracts.WorkflowEvent{
-				WorkflowRunId:  payload.WorkflowRunId.String(),
-				ResourceType:   contracts.ResourceType_RESOURCE_TYPE_STEP_RUN,
-				ResourceId:     payload.TaskRunId.String(),
-				EventType:      contracts.ResourceEventType_RESOURCE_EVENT_TYPE_STREAM,
-				EventTimestamp: timestamppb.New(payload.CreatedAt),
-				EventPayload:   string(payload.Payload),
-				EventIndex:     payload.EventIndex,
-			})
-		}
-	case "workflow-run-finished":
-		payloads := msgqueue.JSONConvert[tasktypes.NotifyFinalizedPayload](payloads)
-
-		for _, payload := range payloads {
-			eventType := contracts.ResourceEventType_RESOURCE_EVENT_TYPE_COMPLETED
-
-			switch payload.Status {
-			case sqlcv1.V1ReadableStatusOlapCANCELLED:
-				eventType = contracts.ResourceEventType_RESOURCE_EVENT_TYPE_CANCELLED
-			case sqlcv1.V1ReadableStatusOlapFAILED:
-				eventType = contracts.ResourceEventType_RESOURCE_EVENT_TYPE_FAILED
-			case sqlcv1.V1ReadableStatusOlapCOMPLETED:
-				eventType = contracts.ResourceEventType_RESOURCE_EVENT_TYPE_COMPLETED
-			}
-
-			workflowEvents = append(workflowEvents, &contracts.WorkflowEvent{
-				WorkflowRunId:  payload.ExternalId.String(),
-				ResourceType:   contracts.ResourceType_RESOURCE_TYPE_WORKFLOW_RUN,
-				ResourceId:     payload.ExternalId.String(),
-				EventType:      eventType,
-				EventTimestamp: timestamppb.New(time.Now()),
-			})
-		}
-	}
-
-	matches, err := filter(workflowEvents)
-
-	if err != nil {
-		return nil, err
-	}
-
-	matches, err = hangupFunc(matches)
-
-	if err != nil {
-		return nil, err
-	}
-
-	// order matches
-	slices.SortFunc(matches, func(a, b *contracts.WorkflowEvent) int {
-		// anything with a hangup should be last
-		if a.Hangup && !b.Hangup {
-			return 1
-		} else if !a.Hangup && b.Hangup {
-			return -1
-		}
-
-		return sortByEventIndex(a, b)
-	})
-
-	return matches, nil
-}
-
-func (s *DispatcherImpl) isMatchingWorkflowRunV1(msg *msgqueue.Message, acks *workflowRunAcks) ([]uuid.UUID, bool) {
-	switch msg.ID {
-	case "workflow-run-finished":
-		payloads := msgqueue.JSONConvert[tasktypes.NotifyFinalizedPayload](msg.Payloads)
-		res := make([]uuid.UUID, 0)
-
-		for _, payload := range payloads {
-			if acks.hasWorkflowRun(payload.ExternalId) {
-				res = append(res, payload.ExternalId)
-			}
-		}
-
-		if len(res) == 0 {
-			return nil, false
-		}
-
-		return res, true
-	case "workflow-run-finished-candidate":
-		payloads := msgqueue.JSONConvert[tasktypes.CandidateFinalizedPayload](msg.Payloads)
-		res := make([]uuid.UUID, 0)
-
-		for _, payload := range payloads {
-			if acks.hasWorkflowRun(payload.WorkflowRunId) {
-				res = append(res, payload.WorkflowRunId)
-			}
-		}
-
-		if len(res) == 0 {
-			return nil, false
-		}
-
-		return res, true
-	default:
-		return nil, false
-	}
-}