fix deadlock on sink shutdown due to full, undrained channel to which nothing can send

Tofel · Tofel · commit 93af25c0d9f9 · 2026-04-10T17:15:53.000+02:00
diff --git a/system-tests/tests/regression/cre/v2_consensus_regression_test.go b/system-tests/tests/regression/cre/v2_consensus_regression_test.go
@@ -48,9 +48,7 @@ func ConsensusFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, consensus
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	for _, bcOutput := range testEnv.CreEnvironment.Blockchains {
diff --git a/system-tests/tests/regression/cre/v2_evm_regression_test.go b/system-tests/tests/regression/cre/v2_evm_regression_test.go
@@ -202,9 +202,12 @@ func EVMReadFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, evmNegative
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(
+			ctx,
+			server,
+			userLogsCh,
+			baseMessageCh,
+		)
 	})
 
 	for _, bcOutput := range testEnv.CreEnvironment.Blockchains {
@@ -272,9 +275,12 @@ func EVMLogTriggerFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, evmNe
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(
+			ctx,
+			server,
+			userLogsCh,
+			baseMessageCh,
+		)
 	})
 	// drain user logs channel in the background, we are not asserting anything on it
 	t_helpers.IgnoreUserLogs(t.Context(), userLogsCh)
@@ -374,9 +380,12 @@ func EVMWriteFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, evmNegativ
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(
+			ctx,
+			server,
+			userLogsCh,
+			baseMessageCh,
+		)
 	})
 
 	for _, bcOutput := range testEnv.CreEnvironment.Blockchains {
diff --git a/system-tests/tests/regression/cre/v2_http_action_regression_test.go b/system-tests/tests/regression/cre/v2_http_action_regression_test.go
@@ -129,9 +129,7 @@ func HTTPActionFailureTest(t *testing.T, testEnv *ttypes.TestEnvironment, httpAc
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	// Now register and deploy the workflow
diff --git a/system-tests/tests/regression/cre/v2_http_trigger_regression_test.go b/system-tests/tests/regression/cre/v2_http_trigger_regression_test.go
@@ -113,9 +113,7 @@ func HTTPTriggerFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, httpNeg
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 	// drain user logs channel in the background, we are not asserting anything on it
 	t_helpers.IgnoreUserLogs(t.Context(), userLogsCh)
diff --git a/system-tests/tests/smoke/cre/v2_aptos_capability_test.go b/system-tests/tests/smoke/cre/v2_aptos_capability_test.go
@@ -171,9 +171,7 @@ func executeAptosScenarios(t *testing.T, tenv *configuration.TestEnvironment, sc
 				// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 				defer cancel()
-				server.Shutdown(ctx)
-				close(userLogsCh)
-				close(baseMessageCh)
+				t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 			})
 
 			scenario.run(t, scenarioEnv, scenarioAptosChain, userLogsCh, baseMessageCh)
diff --git a/system-tests/tests/smoke/cre/v2_consensus_capability_test.go b/system-tests/tests/smoke/cre/v2_consensus_capability_test.go
@@ -26,9 +26,7 @@ func ExecuteConsensusTest(t *testing.T, testEnv *ttypes.TestEnvironment) {
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	workflowName := t_helpers.UniqueWorkflowName(testEnv, "consensustest")
diff --git a/system-tests/tests/smoke/cre/v2_dontime_test.go b/system-tests/tests/smoke/cre/v2_dontime_test.go
@@ -30,9 +30,7 @@ func ExecuteDonTimeTest(t *testing.T, testEnv *ttypes.TestEnvironment) {
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	testLogger.Info().Msg("Creating Cron workflow configuration file...")
diff --git a/system-tests/tests/smoke/cre/v2_evm_capability_test.go b/system-tests/tests/smoke/cre/v2_evm_capability_test.go
@@ -85,9 +85,7 @@ func ExecuteEVMReadTestForCases(t *testing.T, testEnv *ttypes.TestEnvironment, t
 				// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 				defer cancel()
-				server.Shutdown(ctx)
-				close(userLogsCh)
-				close(baseMessageCh)
+				t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 			})
 
 			for _, bcOutput := range perCaseEnv.CreEnvironment.Blockchains {
@@ -397,9 +395,7 @@ func ExecuteEVMLogTriggerTest(t *testing.T, testEnv *ttypes.TestEnvironment) {
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	for _, bcOutput := range testEnv.CreEnvironment.Blockchains {
diff --git a/system-tests/tests/smoke/cre/v2_grpc_source_test.go b/system-tests/tests/smoke/cre/v2_grpc_source_test.go
@@ -270,8 +270,7 @@ func startWorkflowEventSink(t *testing.T) (context.Context, <-chan proto.Message
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(messageChan)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, messageChan)
 	})
 
 	timeout := 5 * time.Minute
diff --git a/system-tests/tests/smoke/cre/v2_http_action_test.go b/system-tests/tests/smoke/cre/v2_http_action_test.go
@@ -131,9 +131,7 @@ func HTTPActionRegressionTest(t *testing.T, testEnv *ttypes.TestEnvironment, url
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	testLogger.Info().Msg("Waiting for HTTP Action regression workflow to complete...")
@@ -229,9 +227,7 @@ func HTTPActionSuccessTest(t *testing.T, testEnv *ttypes.TestEnvironment, httpAc
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	// Wait for workflow execution to complete and verify success
diff --git a/system-tests/tests/smoke/cre/v2_sharding_test.go b/system-tests/tests/smoke/cre/v2_sharding_test.go
@@ -321,9 +321,7 @@ func validateShardingScaleScenario(t *testing.T, testEnv *ttypes.TestEnvironment
 		// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		server.Shutdown(ctx)
-		close(userLogsCh)
-		close(baseMessageCh)
+		t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh)
 	})
 
 	execTimeout := 3 * time.Minute
diff --git a/system-tests/tests/smoke/cre/v2_vault_don_test.go b/system-tests/tests/smoke/cre/v2_vault_don_test.go
@@ -100,9 +100,7 @@ func ExecuteVaultTest(t *testing.T, testEnv *ttypes.TestEnvironment) {
 			// can't use t.Context() here because it will have been cancelled before the cleanup function is called
 			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 			defer cancel()
-			sink.Shutdown(ctx)
-			close(ulCh)
-			close(bmCh)
+			t_helpers.ShutdownChipSinkWithDrain(ctx, sink, ulCh, bmCh)
 		})
 		namespaces := []string{"main", "alt"}
 
diff --git a/system-tests/tests/test-helpers/chip_testsink_helpers.go b/system-tests/tests/test-helpers/chip_testsink_helpers.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
 	"sync"
 	"testing"
@@ -53,6 +54,145 @@ func (s *registeredChipSink) Shutdown(ctx context.Context) {
 	}
 }
 
+// StartChannelDrainers starts one goroutine per channel and drains messages until stop is called
+// or the channel is closed. This is useful during teardown to avoid producer goroutines blocking
+// on full channels while infrastructure is shutting down.
+func StartChannelDrainers[T any](channels ...<-chan T) func() {
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+
+	for _, ch := range channels {
+		if ch == nil {
+			continue
+		}
+
+		wg.Add(1)
+		go func(ch <-chan T) {
+			defer wg.Done()
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case _, ok := <-ch:
+					if !ok {
+						return
+					}
+				}
+			}
+		}(ch)
+	}
+
+	return func() {
+		cancel()
+		wg.Wait()
+	}
+}
+
+// CloseChannels closes all non-nil channels in order.
+func CloseChannels[T any](channels ...chan T) {
+	for _, ch := range channels {
+		if ch != nil {
+			close(ch)
+		}
+	}
+}
+
+// collectChannels flattens channel arguments and slices/arrays of channels into a single list.
+func collectChannels(args ...any) []reflect.Value {
+	channels := make([]reflect.Value, 0, len(args))
+	for _, arg := range args {
+		v := reflect.ValueOf(arg)
+		if !v.IsValid() {
+			continue
+		}
+
+		switch v.Kind() {
+		case reflect.Chan:
+			channels = append(channels, v)
+		case reflect.Slice, reflect.Array:
+			for i := 0; i < v.Len(); i++ {
+				elem := v.Index(i)
+				if elem.IsValid() && elem.Kind() == reflect.Chan {
+					channels = append(channels, elem)
+				}
+			}
+		default:
+			framework.L.Warn().Msgf("unsupported arg to ShutdownChipSinkWithDrainAndClose: %T", arg)
+		}
+	}
+	return channels
+}
+
+// startReflectChannelDrainers drains channels until they are closed or stop() is called.
+func startReflectChannelDrainers(channels []reflect.Value) func() {
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+
+	for _, ch := range channels {
+		if !ch.IsValid() || ch.Kind() != reflect.Chan || ch.IsNil() {
+			continue
+		}
+
+		wg.Add(1)
+		go func(ch reflect.Value) {
+			defer wg.Done()
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+
+				// Recv blocks until an element is available or the channel closes.
+				_, ok := ch.Recv()
+				if !ok {
+					return
+				}
+			}
+		}(ch)
+	}
+
+	return func() {
+		cancel()
+		wg.Wait()
+	}
+}
+
+func closeReflectChannels(channels []reflect.Value) {
+	for _, ch := range channels {
+		if !ch.IsValid() || ch.Kind() != reflect.Chan || ch.IsNil() {
+			continue
+		}
+		// Only close channels that have send capability.
+		if ch.Type().ChanDir()&reflect.SendDir == 0 {
+			continue
+		}
+		func(ch reflect.Value) {
+			defer func() { _ = recover() }()
+			ch.Close()
+		}(ch)
+	}
+}
+
+// ShutdownChipSinkWithDrain performs a safer sink teardown for tests.
+// It starts background drainers for provided channels, shuts down the sink,
+// stops drainers, and then closes channels.
+func ShutdownChipSinkWithDrain(
+	shutdownCtx context.Context,
+	sink ChipSink,
+	channels ...any,
+) {
+	flattenedChannels := collectChannels(channels...)
+	stopDrainers := startReflectChannelDrainers(flattenedChannels)
+
+	if sink != nil {
+		sink.Shutdown(shutdownCtx)
+	}
+
+	stopDrainers()
+	closeReflectChannels(flattenedChannels)
+}
+
 type baseMessageWatchCfg struct {
 	workflowID string
 	labelEq    map[string]string
diff --git a/system-tests/tests/test-helpers/workflow_event_observer.go b/system-tests/tests/test-helpers/workflow_event_observer.go
@@ -56,9 +56,7 @@ func NewSinkWorkflowEventObserver(t *testing.T, testLogger zerolog.Logger) Workf
 		baseMessageCh: baseMessageCh,
 		errCh:         make(chan error),
 		shutdownFn: func(ctx context.Context) {
-			sink.Shutdown(ctx)
-			close(userLogsCh)
-			close(baseMessageCh)
+			ShutdownChipSinkWithDrain(ctx, sink, userLogsCh, baseMessageCh)
 		},
 	}
 }
