[Ouriel]

Summary

Refs #297 — Ensures AddTagToWorkflow/RemoveTagFromWorkflow messages always use the optimized bulk storage path, even when other message types are in the same batch.

Problem

batchProcess() has a fast-path (batchProcessAddRemoveOnly) that does a single bulk Redis read + single bulk Redis write (2 storage calls regardless of batch size). However, this fast-path only activates when all messages in the batch are add/remove.

When any other message type is present (e.g., ContinueWorkflowTagFanout, DispatchWorkflowByCustomId), the entire batch falls into the sequential fallback path where each message hits storage individually — O(N) calls.

Production metrics showed AddTagToWorkflow latency went from sub-ms to 6.5s because these messages were stuck in mixed batches.

Fix

Partition the batch: add/remove messages always go through the bulk path, other messages are processed separately per tag.

val (addRemove, others) = messages.partition { (message, _) ->
    message is AddTagToWorkflow || message is RemoveTagFromWorkflow
}
if (addRemove.isNotEmpty()) batchProcessAddRemoveOnly(addRemove)
if (others.isNotEmpty()) { /* process per tag */ }

What was tested

All 10 existing tests pass, including the batch equivalence test that verifies batch and one-by-one processing produce identical results.

[geomagilles]

Redundant with 0.18.4