Function TestSharedVariableTimeoutHandling

data/shared_state_variable_test.go:989–1197

TestSharedVariableTimeoutHandling verifies that when the remote backend is down, the shared variable operations fail fast and leave enough time for the actual operation.

(t *testing.T)

// TestSharedVariableTimeoutHandling verifies that when the remote backend is down,
// the shared variable operations fail fast and leave enough time for the actual operation.
func TestSharedVariableTimeoutHandling(t *testing.T) {
	t.Run("operations respect timeout when remote is down", func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()

		// Create a connector that simulates a broken Redis connection
		connector := NewMockConnector("broken")
		registry := &sharedStateRegistry{
			appCtx:          ctx,
			logger:          &log.Logger,
			clusterKey:      "test",
			connector:       connector,
			fallbackTimeout: 1 * time.Second,
			lockTtl:         30 * time.Second,
			lockMaxWait:     200 * time.Millisecond,
			updateMaxWait:   200 * time.Millisecond,
			initializer:     util.NewInitializer(ctx, &log.Logger, nil),
		}

		// Setup mock to simulate timeout on Lock
		connector.On("Lock", mock.Anything, mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				ctx := args.Get(0).(context.Context)
				// Wait for context to timeout
				<-ctx.Done()
			}).
			Return(nil, context.DeadlineExceeded)

		// Background push publishes best-effort for fast propagation.
		connector.On("PublishCounterInt64", mock.Anything, "test/timeout-counter", mock.Anything).Return(nil).Maybe()

		// Setup mock for Get to return not found initially
		connector.On("Get", mock.Anything, ConnectorMainIndex, "test/timeout-counter", "value", nil).
			Return(nil, common.NewErrRecordNotFound("test/timeout-counter", "value", "mock"))

		// Setup mock for WatchCounterInt64
		updates := make(chan CounterInt64State, 10)
		cleanup := func() { close(updates) }
		connector.On("WatchCounterInt64", mock.Anything, "test/timeout-counter").
			Return(updates, cleanup, nil)

		// Get a counter variable
		counter := registry.GetCounterInt64("timeout-counter", 100).(*counterInt64)
		time.Sleep(100 * time.Millisecond) // Allow initialization

		// Test TryUpdateIfStale with a parent context that has enough time for lock wait + operations
		operationCtx, operationCancel := context.WithTimeout(ctx, 3*time.Second)
		defer operationCancel()

		executionCount := atomic.Int32{}
		start := time.Now()

		value, err := counter.TryUpdateIfStale(operationCtx, 100*time.Millisecond, func(ctx context.Context) (int64, error) {
			executionCount.Add(1)
			// This simulates the actual operation (e.g., fetching block number)
			select {
			case <-ctx.Done():
				return 0, ctx.Err()