Function testQueueResched

preconf-share/simqueue/queue_test.go:386–444

(t *testing.T, r queueRunner, worker *testWorker)

}

func testQueueResched(t *testing.T, r queueRunner, worker *testWorker) {
	t.Helper()
	ctx := context.Background()
	r.queue.Config.MaxRetries = 3
	defer func() {
		r.queue.Config.MaxRetries = DefaultQueueConfig.MaxRetries
	}()

	var (
		callCount        = 0
		shouldReschedule = true
		mu               sync.Mutex
	)
	processErr := func(ctx context.Context, data []byte, info QueueItemInfo) error {
		mu.Lock()
		defer mu.Unlock()
		callCount++
		if shouldReschedule {
			return ErrProcessScheduleNextBlock
		} else {
			return worker.processOk(ctx, data, info)
		}
	}

	err := r.queue.UpdateBlock(7)
	require.NoError(t, err)

	r.startProcessLoop(ctx, []ProcessFunc{processErr})

	err = r.queue.Push(ctx, []byte("test-error-reschedule"), false, 8, 10)
	require.NoError(t, err)

	// it should fail for current block 7, but reschedule for block 8
	require.Nil(t, worker.nextProcessed(processTimeout))
	err = r.queue.UpdateBlock(8)
	require.NoError(t, err)

	// it should fail for current block 8, but reschedule for block 9, where it should succeed
	time.Sleep(processTimeout)

	mu.Lock()
	shouldReschedule = false
	mu.Unlock()

	require.Nil(t, worker.nextProcessed(processTimeout))
	err = r.queue.UpdateBlock(9)
	require.NoError(t, err)

	require.Equal(t, "test-error-reschedule", string(worker.nextProcessed(processTimeout)))

	require.Nil(t, worker.nextProcessed(processTimeout))

	// 2 failures + 1 success
	mu.Lock()
	require.Equal(t, 3, callCount)
	mu.Unlock()

	r.stopProcessLoop()