anyqueue_test.go

package anyqueue_test

import (
	"context"
	"fmt"
	"log/slog"
	"sync/atomic"
	"testing"
	"time"

	"github.com/pkg/errors"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/talon-one/anyqueue"
)

// TestRun_ProcessesBatch covers the happy path: every item is handled, its
// result recorded, and the batch committed once.
func TestRun_ProcessesBatch(t *testing.T) {
	t.Parallel()

	topic := testTopic()
	committed := make(chan struct{})

	full := anyqueue.NewMockBatch[int, string](t)
	full.EXPECT().Items().Return([]int{1, 2, 3})
	full.EXPECT().Record(mock.Anything, 1, "ok-1", nil).Return(nil).Once()
	full.EXPECT().Record(mock.Anything, 2, "ok-2", nil).Return(nil).Once()
	full.EXPECT().Record(mock.Anything, 3, "ok-3", nil).Return(nil).Once()
	full.EXPECT().Commit(mock.Anything).Return(nil).Run(func(context.Context) { close(committed) }).Once()

	store := anyqueue.NewMockStore[fakeTopic, int, string](t)
	store.EXPECT().ListTopics(mock.Anything).Return([]fakeTopic{topic}, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(full, nil).Once()
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(drainedBatch(t), nil).Maybe()

	metrics := anyqueue.NewMockMetrics(t)
	metrics.EXPECT().ItemProcessed(mock.Anything, true).Times(3)
	metrics.EXPECT().BatchProcessed(mock.Anything).Maybe()

	handler := func(_ context.Context, _ *slog.Logger, item int) (string, error) {
		return fmt.Sprintf("ok-%d", item), nil
	}

	runEngine(t, store, metrics, handler, committed)
}

// TestRun_EmptyBatchRollsBack covers the no-work path: an empty batch is rolled
// back, never committed, and the handler is never invoked.
func TestRun_EmptyBatchRollsBack(t *testing.T) {
	t.Parallel()

	topic := testTopic()
	rolled := make(chan struct{}, 1)

	empty := anyqueue.NewMockBatch[int, string](t)
	empty.EXPECT().Items().Return(nil)
	empty.EXPECT().Rollback(mock.Anything).Return(nil).Run(func(context.Context) {
		select {
		case rolled <- struct{}{}:
		default:
		}
	})

	store := anyqueue.NewMockStore[fakeTopic, int, string](t)
	store.EXPECT().ListTopics(mock.Anything).Return([]fakeTopic{topic}, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(empty, nil)

	var handlerCalls atomic.Int32
	handler := func(_ context.Context, _ *slog.Logger, _ int) (string, error) {
		handlerCalls.Add(1)

		return "", nil
	}

	// No metric calls are expected for empty batches.
	runEngine(t, store, anyqueue.NewMockMetrics(t), handler, rolled)

	assert.Zero(t, handlerCalls.Load())
}

// TestRun_ConcurrentWorkers verifies a batch is fanned out across WorkerCount
// workers: with 3 workers the 3 items must be in flight simultaneously. Each
// handler blocks until all three have entered, so a smaller fan-out would
// deadlock the barrier and time out.
func TestRun_ConcurrentWorkers(t *testing.T) {
	t.Parallel()

	const workers = 3
	topic := fakeTopic{id: 1, cfg: anyqueue.TopicConfig{WorkerCount: workers, PollInterval: time.Millisecond}}

	entered := make(chan struct{}, workers)
	release := make(chan struct{})
	committed := make(chan struct{})

	full := anyqueue.NewMockBatch[int, string](t)
	full.EXPECT().Items().Return([]int{1, 2, 3})
	full.EXPECT().Record(mock.Anything, mock.Anything, mock.Anything, nil).Return(nil).Times(workers)
	full.EXPECT().Commit(mock.Anything).Return(nil).Run(func(context.Context) { close(committed) }).Once()

	store := anyqueue.NewMockStore[fakeTopic, int, string](t)
	store.EXPECT().ListTopics(mock.Anything).Return([]fakeTopic{topic}, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(full, nil).Once()
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(drainedBatch(t), nil).Maybe()

	metrics := anyqueue.NewMockMetrics(t)
	metrics.EXPECT().ItemProcessed(mock.Anything, true).Times(workers)
	metrics.EXPECT().BatchProcessed(mock.Anything).Maybe()

	var handler anyqueue.Handler[int, string] = func(_ context.Context, _ *slog.Logger, _ int) (string, error) {
		entered <- struct{}{}
		<-release

		return "ok", nil
	}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() {
		done <- anyqueue.Run[fakeTopic, int, string](ctx, discardLogger(), metrics, anyqueue.Config{}, store, handler)
	}()

	// All three handlers must be running concurrently before any is released.
	for i := range workers {
		select {
		case <-entered:
		case <-time.After(2 * time.Second):
			cancel()
			t.Fatalf("only %d/%d workers ran concurrently", i, workers)
		}
	}
	close(release)

	select {
	case <-committed:
	case <-time.After(2 * time.Second):
		cancel()
		t.Fatal("batch was not committed")
	}

	cancel()
	select {
	case err := <-done:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("engine did not stop after cancel")
	}
}

// TestRun_ShutdownDuringHandlerDoesNotHang covers graceful shutdown while a
// batch is in flight: canceling the context must stop the engine cleanly and
// never hang the topic goroutine. (Regression guard for the abort-mid-batch
// deadlock where a worker could block forever on its workDone send.)
func TestRun_ShutdownDuringHandlerDoesNotHang(t *testing.T) {
	t.Parallel()

	topic := fakeTopic{id: 1, cfg: anyqueue.TopicConfig{WorkerCount: 1, PollInterval: time.Millisecond}}
	entered := make(chan struct{}, 1)

	full := anyqueue.NewMockBatch[int, string](t)
	full.EXPECT().Items().Return([]int{1, 2})
	// Either outcome is acceptable depending on scheduling; the point is a clean stop.
	full.EXPECT().Record(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
	full.EXPECT().Commit(mock.Anything).Return(nil).Maybe()
	full.EXPECT().Rollback(mock.Anything).Return(nil).Maybe()

	store := anyqueue.NewMockStore[fakeTopic, int, string](t)
	store.EXPECT().ListTopics(mock.Anything).Return([]fakeTopic{topic}, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(full, nil).Once()
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(drainedBatch(t), nil).Maybe()

	metrics := anyqueue.NewMockMetrics(t)
	metrics.EXPECT().ItemProcessed(mock.Anything, mock.Anything).Maybe()
	metrics.EXPECT().BatchProcessed(mock.Anything).Maybe()

	// The handler honors its context so cancellation unblocks it.
	var handler anyqueue.Handler[int, string] = func(ctx context.Context, _ *slog.Logger, _ int) (string, error) {
		select {
		case entered <- struct{}{}:
		default:
		}
		<-ctx.Done()

		return "", ctx.Err()
	}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() {
		done <- anyqueue.Run[fakeTopic, int, string](ctx, discardLogger(), metrics, anyqueue.Config{}, store, handler)
	}()

	// Cancel only once a handler is actually in flight (batch is being processed).
	select {
	case <-entered:
	case <-time.After(2 * time.Second):
		cancel()
		t.Fatal("handler never ran")
	}
	cancel()

	select {
	case err := <-done:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("engine hung on shutdown during in-flight batch")
	}
}

// TestRun_TopicsAreIndependent verifies a slow topic does not block others:
// topic 1's handler blocks indefinitely, yet topic 2's item must still be
// processed. Each topic runs in its own goroutine, so topic 2 progresses while
// topic 1 is stuck.
func TestRun_TopicsAreIndependent(t *testing.T) {
	t.Parallel()

	topic1 := fakeTopic{id: 1, cfg: anyqueue.TopicConfig{WorkerCount: 1, PollInterval: time.Millisecond}}
	topic2 := fakeTopic{id: 2, cfg: anyqueue.TopicConfig{WorkerCount: 1, PollInterval: time.Millisecond}}

	blocked := make(chan struct{}, 1)   // topic 1's handler has entered (and is stuck)
	processed := make(chan struct{}, 1) // topic 2's item was handled

	// topic 1 yields one item whose handler blocks until shutdown.
	batch1 := anyqueue.NewMockBatch[int, string](t)
	batch1.EXPECT().Items().Return([]int{1})
	batch1.EXPECT().Record(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
	batch1.EXPECT().Commit(mock.Anything).Return(nil).Maybe()
	batch1.EXPECT().Rollback(mock.Anything).Return(nil).Maybe()

	// topic 2 yields one item that is handled immediately.
	batch2 := anyqueue.NewMockBatch[int, string](t)
	batch2.EXPECT().Items().Return([]int{2})
	batch2.EXPECT().Record(mock.Anything, 2, "ok", nil).Return(nil).Once()
	batch2.EXPECT().Commit(mock.Anything).Return(nil).Maybe()

	store := anyqueue.NewMockStore[fakeTopic, int, string](t)
	store.EXPECT().ListTopics(mock.Anything).Return([]fakeTopic{topic1, topic2}, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic1).Return(batch1, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic2).Return(batch2, nil).Once()
	store.EXPECT().BeginBatch(mock.Anything, topic2).Return(drainedBatch(t), nil).Maybe()

	metrics := anyqueue.NewMockMetrics(t)
	metrics.EXPECT().ItemProcessed(mock.Anything, mock.Anything).Maybe()
	metrics.EXPECT().BatchProcessed(mock.Anything).Maybe()

	var handler anyqueue.Handler[int, string] = func(ctx context.Context, _ *slog.Logger, item int) (string, error) {
		if item == 1 {
			select {
			case blocked <- struct{}{}:
			default:
			}
			<-ctx.Done() // stay busy until shutdown

			return "", ctx.Err()
		}
		select {
		case processed <- struct{}{}:
		default:
		}

		return "ok", nil
	}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() {
		done <- anyqueue.Run[fakeTopic, int, string](ctx, discardLogger(), metrics, anyqueue.Config{}, store, handler)
	}()

	// topic 1 is stuck...
	select {
	case <-blocked:
	case <-time.After(2 * time.Second):
		cancel()
		t.Fatal("topic 1 handler never ran")
	}
	// ...yet topic 2 must still be processed.
	select {
	case <-processed:
	case <-time.After(2 * time.Second):
		cancel()
		t.Fatal("topic 2 was blocked by topic 1")
	}

	cancel()
	select {
	case err := <-done:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("engine did not stop after cancel")
	}
}

// TestRun_TopicDiscovery covers the discovery loop: a goroutine is spun up for
// each new topic and torn down when a topic disappears or the engine stops.
func TestRun_TopicDiscovery(t *testing.T) {
	t.Parallel()

	topic1 := fakeTopic{id: 1, cfg: anyqueue.TopicConfig{WorkerCount: 1, PollInterval: time.Millisecond}}
	topic2 := fakeTopic{id: 2, cfg: anyqueue.TopicConfig{WorkerCount: 1, PollInterval: time.Millisecond}}

	store := newDiscoveryStore(topic1)

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() {
		// A short sync interval makes topic add/remove observable quickly.
		cfg := anyqueue.Config{TopicSyncInterval: 5 * time.Millisecond}
		handler := func(_ context.Context, _ *slog.Logger, _ int) (string, error) { return "", nil }
		done <- anyqueue.Run[fakeTopic, int, string](ctx, discardLogger(), anyqueue.NewMockMetrics(t), cfg, store, handler)
	}()

	// topic1 is discovered at startup.
	waitForTopicID(t, store.started, 1)

	// Adding topic2 spins up its goroutine.
	store.setTopics(topic1, topic2)
	waitForTopicID(t, store.started, 2)

	// Removing topic1 tears down its goroutine (its context is cancelled).
	store.setTopics(topic2)
	waitForTopicID(t, store.stopped, 1)

	// Shutdown tears down the remaining topic.
	cancel()
	waitForTopicID(t, store.stopped, 2)

	select {
	case err := <-done:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("engine did not stop after cancel")
	}
}

// TestRun_HandlerErrorRecorded covers failure propagation: the handler's error
// reaches Record, the item is metered as unsuccessful, and the batch still commits.
func TestRun_HandlerErrorRecorded(t *testing.T) {
	t.Parallel()

	topic := testTopic()
	errBoom := errors.New("boom")
	committed := make(chan struct{})

	full := anyqueue.NewMockBatch[int, string](t)
	full.EXPECT().Items().Return([]int{42})
	full.EXPECT().Record(mock.Anything, 42, "", errBoom).Return(nil).Once()
	full.EXPECT().Commit(mock.Anything).Return(nil).Run(func(context.Context) { close(committed) }).Once()

	store := anyqueue.NewMockStore[fakeTopic, int, string](t)
	store.EXPECT().ListTopics(mock.Anything).Return([]fakeTopic{topic}, nil)
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(full, nil).Once()
	store.EXPECT().BeginBatch(mock.Anything, topic).Return(drainedBatch(t), nil).Maybe()

	metrics := anyqueue.NewMockMetrics(t)
	metrics.EXPECT().ItemProcessed(mock.Anything, false).Once()
	metrics.EXPECT().BatchProcessed(mock.Anything).Maybe()

	handler := func(_ context.Context, _ *slog.Logger, _ int) (string, error) {
		return "", errBoom
	}

	runEngine(t, store, metrics, handler, committed)
}