Add Sometimes assertions (#8571)

## What changed?

WISOTT

## Why?

Sometimes assertions help our internal testing platform to identify
interesting test cases.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks

Performance overhead in production; but it should be minimal as the
assertions are not placed on the hot path but on error cases.

Author: stephanos

common/softassert/softassert.go

@@ -16,6 +16,10 @@ import (
 	"go.temporal.io/server/common/log/tag"
 )
 
+type sometimesLogger struct {
+	logger log.Logger
+}
+
 // That performs a soft assertion by logging an error if the given condition is false.
 // It is meant to indicate a condition is always expected to be true.
 // Returns true if the condition is met, otherwise false.
@@ -44,27 +48,42 @@ func Fail(logger log.Logger, staticMessage string, tags ...tag.Tag) {
 	logger.Error("failed assertion: "+staticMessage, append([]tag.Tag{tag.FailedAssertion}, tags...)...)
 }
 
-// ThatSometimes is used to conditionally log a debug message of a noteworthy but non-problematic event.
+// Sometimes is used to log a message of a noteworthy but non-problematic event.
+//
+// Example:
+// softassert.Sometimes(logger).Warn("termination event", tag.NewStringTag("state", object.state))
+func Sometimes(logger log.Logger) *sometimesLogger {
+	return &sometimesLogger{logger: logger}
+}
+
+// Debug logs a message at debug level.
 //
 // `staticMessage` is expected to be a static string to help with grouping and searching logs.
 // Dynamic information should be passed via `tags`.
+func (s *sometimesLogger) Debug(staticMessage string, tags ...tag.Tag) {
+	s.logger.Debug(staticMessage, tags...)
+}
+
+// Info logs a message at info level.
 //
-// Example:
-// softassert.ThatSometimes(logger, object.state == "terminated", "termination event", tag.NewStringTag("state", object.state))
-func ThatSometimes(logger log.Logger, condition bool, staticMessage string, tags ...tag.Tag) bool {
-	if !condition {
-		logger.Debug(staticMessage, tags...)
-	}
-	return condition
+// `staticMessage` is expected to be a static string to help with grouping and searching logs.
+// Dynamic information should be passed via `tags`.
+func (s *sometimesLogger) Info(staticMessage string, tags ...tag.Tag) {
+	s.logger.Info(staticMessage, tags...)
 }
 
-// Sometimes is used to log a debug message of a noteworthy but non-problematic event.
+// Warn logs a message at warn level.
 //
 // `staticMessage` is expected to be a static string to help with grouping and searching logs.
 // Dynamic information should be passed via `tags`.
+func (s *sometimesLogger) Warn(staticMessage string, tags ...tag.Tag) {
+	s.logger.Warn(staticMessage, tags...)
+}
+
+// Error logs a message at error level.
 //
-// Example:
-// softassert.Sometimes(logger, "termination event", tag.NewStringTag("state", object.state))
-func Sometimes(logger log.Logger, staticMessage string, tags ...tag.Tag) {
-	logger.Debug(staticMessage, tags...)
+// `staticMessage` is expected to be a static string to help with grouping and searching logs.
+// Dynamic information should be passed via `tags`.
+func (s *sometimesLogger) Error(staticMessage string, tags ...tag.Tag) {
+	s.logger.Error(staticMessage, tags...)
 }

service/history/api/respondworkflowtaskcompleted/api.go

@@ -30,6 +30,7 @@ import (
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/searchattribute"
+	"go.temporal.io/server/common/softassert"
 	"go.temporal.io/server/common/tasktoken"
 	"go.temporal.io/server/common/worker_versioning"
 	"go.temporal.io/server/service/history/api"
@@ -129,6 +130,12 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 				metrics.StaleMutableStateCounter.With(handler.metricsHandler).Record(
 					1,
 					metrics.OperationTag(metrics.HistoryRespondWorkflowTaskCompletedScope))
+				softassert.Sometimes(handler.logger).Debug("stale mutable state detected",
+					tag.WorkflowID(token.GetWorkflowId()),
+					tag.WorkflowRunID(token.GetRunId()),
+					tag.WorkflowScheduledEventID(token.GetScheduledEventId()),
+					tag.NewInt64("mutable-state-next-event-id", mutableState.GetNextEventID()),
+				)
 				return false
 			}
 			return true
@@ -166,7 +173,7 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 			// This is NOT 100% bulletproof solution because this write operation may also fail.
 			// TODO: remove this call when GetWorkflowExecutionHistory includes speculative WFT events.
 			if clearStickyErr := handler.clearStickyTaskQueue(ctx, workflowLease.GetContext()); clearStickyErr != nil {
-				handler.logger.Error("Failed to clear stickiness after speculative workflow task failed to complete.",
+				softassert.Sometimes(handler.logger).Error("Failed to clear stickiness after speculative workflow task failed to complete.",
 					tag.NewErrorTag("clear-sticky-error", clearStickyErr),
 					tag.Error(retError),
 					tag.WorkflowID(token.GetWorkflowId()),
@@ -225,7 +232,7 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 		if retError != nil {
 			cancelled := effects.Cancel(ctx)
 			if cancelled {
-				handler.logger.Info("Canceled effects due to error.",
+				softassert.Sometimes(handler.logger).Info("Canceled effects due to error",
 					tag.Error(retError),
 					tag.WorkflowID(token.GetWorkflowId()),
 					tag.WorkflowRunID(token.GetRunId()),
@@ -284,6 +291,11 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 				metrics.NamespaceTag(nsName),
 			)
 			metrics.WorkflowTaskHeartbeatTimeoutCounter.With(scope).Record(1)
+			softassert.Sometimes(handler.logger).Debug("workflow task heartbeat timed out",
+				tag.WorkflowNamespaceID(nsName),
+				tag.WorkflowID(token.GetWorkflowId()),
+				tag.WorkflowRunID(token.GetRunId()),
+			)
 			completedEvent, err = ms.AddWorkflowTaskTimedOutEvent(currentWorkflowTask)
 			if err != nil {
 				return nil, err
@@ -338,6 +350,12 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 	//   and admitted updates are lost. Uncomment this check when durable admitted is implemented
 	//   or updates stay in the registry after WFT is failed.
 	hasBufferedEventsOrMessages := ms.HasBufferedEvents() // || updateRegistry.HasOutgoingMessages(false)
+	if hasBufferedEventsOrMessages {
+		softassert.Sometimes(handler.logger).Debug("workflow has buffered events/messages",
+			tag.WorkflowID(token.GetWorkflowId()),
+			tag.WorkflowRunID(token.GetRunId()),
+		)
+	}
 	if err := namespaceEntry.VerifyBinaryChecksum(request.GetBinaryChecksum()); err != nil {
 		wtFailedCause = newWorkflowTaskFailedCause(
 			enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_BINARY,
@@ -447,11 +465,14 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 			metrics.FailureTag(wtFailedCause.failedCause.String()),
 			metrics.FirstAttemptTag(currentWorkflowTask.Attempt),
 		)
-		handler.logger.Info("Failing the workflow task.",
+		softassert.Sometimes(handler.logger).Info("Failing the workflow task.",
 			tag.Value(wtFailedCause.Message()),
 			tag.WorkflowID(token.GetWorkflowId()),
 			tag.WorkflowRunID(token.GetRunId()),
-			tag.WorkflowNamespaceID(namespaceEntry.ID().String()))
+			tag.WorkflowNamespaceID(namespaceEntry.ID().String()),
+			tag.Attempt(currentWorkflowTask.Attempt),
+			tag.Cause(wtFailedCause.failedCause.String()),
+		)
 		if currentWorkflowTask.Attempt > 1 && wtFailedCause.failedCause != enumspb.WORKFLOW_TASK_FAILED_CAUSE_UNHANDLED_COMMAND {
 			// drop this workflow task if it keeps failing. This will cause the workflow task to timeout and get retried after timeout.
 			return nil, serviceerror.NewInvalidArgument(wtFailedCause.Message())
@@ -620,6 +641,12 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 		// if updateErr resulted in TransactionSizeLimitError then fail workflow
 		switch updateErr.(type) {
 		case *persistence.TransactionSizeLimitError:
+			softassert.Sometimes(handler.logger).Debug("workflow terminated due to size limit",
+				tag.WorkflowID(token.GetWorkflowId()),
+				tag.WorkflowRunID(token.GetRunId()),
+				tag.Error(updateErr),
+			)
+
 			// must reload mutable state because the first call to updateWorkflowExecutionWithContext or continueAsNewWorkflowExecution
 			// clears mutable state if error is returned
 			ms, err = weContext.LoadMutableState(ctx, handler.shardContext)

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go

@@ -34,6 +34,7 @@ import (
 	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/common/protocol"
 	"go.temporal.io/server/common/searchattribute"
+	"go.temporal.io/server/common/softassert"
 	"go.temporal.io/server/common/tasktoken"
 	"go.temporal.io/server/common/worker_versioning"
 	"go.temporal.io/server/service/history/api"
@@ -239,14 +240,15 @@ func (handler *workflowTaskCompletedHandler) rejectUnprocessedUpdates(
 		handler.effects)
 
 	if len(rejectedUpdateIDs) > 0 {
-		handler.logger.Warn(
+		softassert.Sometimes(handler.logger).Warn(
 			"Workflow task completed w/o processing updates.",
 			tag.WorkflowNamespaceID(wfKey.NamespaceID),
 			tag.WorkflowID(wfKey.WorkflowID),
 			tag.WorkflowRunID(wfKey.RunID),
 			tag.WorkflowEventID(workflowTaskScheduledEventID),
 			tag.NewStringTag("worker-identity", workerIdentity),
 			tag.NewStringsTag("update-ids", rejectedUpdateIDs),
+			tag.NewInt("rejected-count", len(rejectedUpdateIDs)),
 		)
 	}
 
@@ -372,12 +374,21 @@ func (handler *workflowTaskCompletedHandler) handleMessage(
 		}
 		if upd == nil {
 			// Update was not found in the registry and can't be resurrected.
+			softassert.Sometimes(handler.logger).Debug("update lost and cannot be resurrected",
+				tag.WorkflowID(handler.mutableState.GetExecutionInfo().WorkflowId),
+				tag.WorkflowRunID(handler.mutableState.GetExecutionState().RunId),
+				tag.NewStringTag("protocol-instance-id", message.ProtocolInstanceId))
 			return handler.failWorkflowTask(
 				enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_UPDATE_WORKFLOW_EXECUTION_MESSAGE,
 				serviceerror.NewNotFoundf("update %s wasn't found on the server. This is most likely a transient error which will be resolved automatically by retries", message.ProtocolInstanceId))
 		}
 
 		if err := upd.OnProtocolMessage(message, workflow.WithEffects(handler.effects, handler.mutableState)); err != nil {
+			softassert.Sometimes(handler.logger).Debug("update failed",
+				tag.WorkflowID(handler.mutableState.GetExecutionInfo().WorkflowId),
+				tag.WorkflowRunID(handler.mutableState.GetExecutionState().RunId),
+				tag.NewStringTag("protocol-instance-id", message.ProtocolInstanceId),
+				tag.NewStringTag("error", err.Error()))
 			return handler.failWorkflowTaskOnInvalidArgument(
 				enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_UPDATE_WORKFLOW_EXECUTION_MESSAGE, err)
 		}

service/history/queues/dlq_writer.go

@@ -11,6 +11,7 @@ import (
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/common/softassert"
 	"go.temporal.io/server/service/history/tasks"
 )
 
@@ -118,11 +119,12 @@ func (q *DLQWriter) WriteTaskToDLQ(
 	} else {
 		namespaceTag = tag.WorkflowNamespace(string(ns.Name()))
 	}
-	q.logger.Warn("Task enqueued to DLQ",
+	softassert.Sometimes(q.logger).Warn("Task enqueued to DLQ",
 		tag.DLQMessageID(resp.Metadata.ID),
 		tag.SourceCluster(sourceCluster),
 		tag.TargetCluster(targetCluster),
 		tag.TaskType(task.GetType()),
+		tag.NewStringTag("task-category", task.GetCategory().Name()),
 		namespaceTag,
 	)
 	return nil

service/history/queues/executable.go

@@ -28,6 +28,7 @@ import (
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/softassert"
 	ctasks "go.temporal.io/server/common/tasks"
 	"go.temporal.io/server/common/telemetry"
 	"go.temporal.io/server/common/util"
@@ -440,6 +441,7 @@ func (e *executableImpl) isExpectedRetryableError(err error) (isRetryable bool,
 	if errors.As(err, &resourceExhaustedErr) {
 		switch resourceExhaustedErr.Cause { //nolint:exhaustive
 		case enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW:
+			softassert.Sometimes(e.logger).Debug("task throttled due to busy workflow", tag.TaskType(e.GetType()))
 			err = consts.ErrResourceExhaustedBusyWorkflow
 		case enumspb.RESOURCE_EXHAUSTED_CAUSE_APS_LIMIT:
 			err = consts.ErrResourceExhaustedAPSLimit
@@ -482,7 +484,14 @@ func (e *executableImpl) isExpectedRetryableError(err error) (isRetryable bool,
 func (e *executableImpl) isUnexpectedNonRetryableError(err error) bool {
 	var terr MaybeTerminalTaskError
 	if errors.As(err, &terr) {
-		return terr.IsTerminalTaskError()
+		isTerminal := terr.IsTerminalTaskError()
+		if isTerminal {
+			softassert.Sometimes(e.logger).Debug("terminal task error detected",
+				tag.TaskType(e.GetType()),
+				tag.Error(err),
+			)
+		}
+		return isTerminal
 	}
 
 	if _, isDataLoss := err.(*serviceerror.DataLoss); isDataLoss {
@@ -492,8 +501,13 @@ func (e *executableImpl) isUnexpectedNonRetryableError(err error) bool {
 	isInternalError := common.IsInternalError(err)
 	if isInternalError {
 		metrics.TaskInternalErrorCounter.With(e.metricsHandler).Record(1)
+		softassert.Sometimes(e.logger).Debug("internal non-retryable task processing error",
+			tag.TaskType(e.GetType()),
+			tag.Error(err),
+		)
 		// Only DQL/drop when configured to
-		return e.dlqInternalErrors()
+		shouldDLQ := e.dlqInternalErrors()
+		return shouldDLQ
 	}
 
 	return false
@@ -538,11 +552,12 @@ func (e *executableImpl) HandleErr(err error) (retErr error) {
 		tag.Attempt(int32(attempt)),
 		tag.UnexpectedErrorAttempts(int32(e.unexpectedErrorAttempts)),
 		tag.LifeCycleProcessingFailed,
+		tag.NewStringTag("task-category", e.GetCategory().Name()),
 	)
 	if attempt > taskCriticalLogMetricAttempts {
-		logger.Error("Critical error processing task, retrying.", tag.OperationCritical)
+		softassert.Sometimes(logger).Error("Critical error processing task, retrying.", tag.OperationCritical)
 	} else {
-		logger.Warn("Fail to process task")
+		softassert.Sometimes(logger).Warn("Fail to process task")
 	}
 
 	if e.isUnexpectedNonRetryableError(err) {

service/history/replication/executable_task.go

@@ -340,7 +340,7 @@ func (e *ExecutableTaskImpl) Resend(
 ) (bool, error) {
 	remainingAttempt--
 	if remainingAttempt < 0 {
-		e.Logger.Error("resend history attempts exceeded",
+		softassert.Sometimes(e.Logger).Error("resend history attempts exceeded",
 			tag.WorkflowNamespaceID(retryErr.NamespaceId),
 			tag.WorkflowID(retryErr.WorkflowId),
 			tag.WorkflowRunID(retryErr.RunId),
@@ -768,7 +768,7 @@ func (e *ExecutableTaskImpl) MarkPoisonPill() error {
 	taskInfo := e.ReplicationTask().GetRawTaskInfo()
 
 	if e.markPoisonPillAttempts >= MarkPoisonPillMaxAttempts {
-		e.Logger.Error("MarkPoisonPill reached max attempts",
+		softassert.Sometimes(e.Logger).Error("MarkPoisonPill reached max attempts",
 			tag.SourceCluster(e.SourceClusterName()),
 			tag.ReplicationTask(taskInfo),
 		)

service/history/timer_queue_active_task_executor.go

@@ -24,6 +24,7 @@ import (
 	"go.temporal.io/server/common/primitives/timestamp"
 	"go.temporal.io/server/common/priorities"
 	"go.temporal.io/server/common/resource"
+	"go.temporal.io/server/common/softassert"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/deletemanager"
@@ -536,7 +537,7 @@ func (t *timerQueueActiveTaskExecutor) executeActivityRetryTimerTask(
 	}
 
 	if task.Attempt < activityInfo.Attempt || activityInfo.StartedEventId != common.EmptyEventID {
-		t.logger.Info("Duplicate activity retry timer task",
+		softassert.Sometimes(t.logger).Info("Duplicate activity retry timer task",
 			tag.WorkflowID(mutableState.GetExecutionInfo().WorkflowId),
 			tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
 			tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().NamespaceId),
@@ -909,6 +910,12 @@ func (t *timerQueueActiveTaskExecutor) emitTimeoutMetricScopeWithNamespaceTag(
 	case enumspb.TIMEOUT_TYPE_HEARTBEAT:
 		metrics.HeartbeatTimeoutCounter.With(metricsScope).Record(1)
 	}
+
+	softassert.Sometimes(t.logger).Debug("timer queue task timed out",
+		tag.NewStringTag("timer-type", timerType.String()),
+		tag.Operation(operation),
+		tag.Attempt(taskAttempt),
+	)
 }
 
 func (t *timerQueueActiveTaskExecutor) processActivityWorkflowRules(

service/history/workflow/update/update.go

@@ -251,7 +251,7 @@ func (u *Update) abort(
 		return
 	}
 
-	u.instrumentation.countAborted(reason)
+	u.instrumentation.countAborted(u.id, reason)
 	prevState := u.setState(stateProvisionallyAborted)
 
 	effects.OnAfterCommit(func(context.Context) {

service/history/workflow/update/util.go

@@ -61,9 +61,13 @@ func (i *instrumentation) countTooMany() {
 	i.oneOf(metrics.WorkflowExecutionUpdateTooMany.Name())
 }
 
-func (i *instrumentation) countAborted(reason AbortReason) {
+func (i *instrumentation) countAborted(updateID string, reason AbortReason) {
 	i.metrics.Counter(metrics.WorkflowExecutionUpdateAborted.Name()).
 		Record(1, metrics.ReasonTag(metrics.ReasonString(reason.String())))
+	softassert.Sometimes(i.log).Debug("update aborted",
+		tag.NewStringTag("reason", reason.String()),
+		tag.NewStringTag("update-id", updateID),
+	)
 }
 
 func (i *instrumentation) countContinueAsNewSuggestions() {
@@ -99,8 +103,7 @@ func (i *instrumentation) oneOf(counterName string) {
 }
 
 func (i *instrumentation) stateChange(updateID string, from, to state) {
-	softassert.Sometimes(
-		i.log,
+	softassert.Sometimes(i.log).Debug(
 		"update state change",
 		tag.ComponentWorkflowUpdate,
 		tag.NewStringTag("update-id", updateID),