proxy-pool/internal/adapters/redisactivity/runtime_integration_test.go

//go:build integration

package redisactivity

import (
	"context"
	"errors"
	"testing"
	"time"

	"proxy-pool/internal/domain/workerruntime"
)

func TestRedisWorkerRuntimeReplacesSparseCountersAndFencesReports(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*time.Minute,
		testProxy("proxy-a", "192.0.2.10"))
	assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(2*time.Second),
		"proxy-a", "worker-a", time.Minute)
	if err != nil {
		t.Fatalf("Assign(): %v", err)
	}
	session := workerruntime.Session{
		WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a",
		AckedSnapshotVersion: 3, AckedOwnershipEpoch: assignment.Epoch,
	}
	if err := fixture.Adapter.ReplaceSession(context.Background(), session, time.Minute); err != nil {
		t.Fatalf("ReplaceSession(): %v", err)
	}
	report := workerruntime.Report{
		WorkerID: "worker-a", SessionID: "session-a", Sequence: 2,
		SnapshotVersion: 3, OwnershipEpoch: assignment.Epoch, ObservedAt: now,
		Counters: []workerruntime.Counter{{ProxyID: "proxy-a", Active: 2, Reserved: 1}},
	}
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), report, time.Minute); err != nil {
		t.Fatalf("ReplaceRuntime(first): %v", err)
	}
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), report, time.Minute); err != nil {
		t.Fatalf("ReplaceRuntime(replay): %v", err)
	}
	conflict := report
	conflict.Counters = []workerruntime.Counter{{ProxyID: "proxy-a", Active: 3}}
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), conflict, time.Minute); !errors.Is(err, workerruntime.ErrConflictingReport) {
		t.Fatalf("ReplaceRuntime(conflict) error = %v", err)
	}
	stale := report
	stale.Sequence = 1
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), stale, time.Minute); !errors.Is(err, workerruntime.ErrStaleReport) {
		t.Fatalf("ReplaceRuntime(stale) error = %v", err)
	}

	query := []workerruntime.OwnedProxy{{
		ProxyID: "proxy-a", WorkerID: "worker-a", OwnershipEpoch: assignment.Epoch,
	}}
	got, err := fixture.Adapter.ReadRuntime(context.Background(), query)
	if err != nil || len(got) != 1 || got[0] != (workerruntime.Snapshot{
		ProxyID: "proxy-a", Active: 2, Reserved: 1, Fresh: true,
	}) {
		t.Fatalf("ReadRuntime(first) = %+v, %v", got, err)
	}
	report.Sequence = 3
	report.Counters = nil
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), report, time.Minute); err != nil {
		t.Fatalf("ReplaceRuntime(empty): %v", err)
	}
	got, err = fixture.Adapter.ReadRuntime(context.Background(), query)
	if err != nil || len(got) != 1 || got[0] != (workerruntime.Snapshot{ProxyID: "proxy-a", Fresh: true}) {
		t.Fatalf("ReadRuntime(empty) = %+v, %v", got, err)
	}

	if err := fixture.Adapter.ReplaceSession(context.Background(), workerruntime.Session{
		WorkerID: "worker-a", InstanceID: "instance-b", SessionID: "session-b",
		AckedSnapshotVersion: 4, AckedOwnershipEpoch: assignment.Epoch + 1,
	}, time.Minute); err != nil {
		t.Fatalf("ReplaceSession(new): %v", err)
	}
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, workerruntime.ErrStaleSession) {
		t.Fatalf("ReplaceRuntime(old session) error = %v", err)
	}
}

func TestRedisWorkerRuntimeExpiresFailClosed(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*time.Minute,
		testProxy("proxy-a", "192.0.2.10"))
	assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(2*time.Second),
		"proxy-a", "worker-a", time.Minute)
	if err != nil {
		t.Fatalf("Assign(): %v", err)
	}
	if err := fixture.Adapter.ReplaceSession(context.Background(), workerruntime.Session{
		WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a",
		AckedSnapshotVersion: 1, AckedOwnershipEpoch: assignment.Epoch,
	}, 100*time.Millisecond); err != nil {
		t.Fatalf("ReplaceSession(): %v", err)
	}
	if err := fixture.Adapter.ReplaceRuntime(context.Background(), workerruntime.Report{
		WorkerID: "worker-a", SessionID: "session-a", Sequence: 1,
		SnapshotVersion: 1, OwnershipEpoch: assignment.Epoch, ObservedAt: now,
	}, 100*time.Millisecond); err != nil {
		t.Fatalf("ReplaceRuntime(): %v", err)
	}
	time.Sleep(150 * time.Millisecond)
	got, err := fixture.Adapter.ReadRuntime(context.Background(), []workerruntime.OwnedProxy{{
		ProxyID: "proxy-a", WorkerID: "worker-a", OwnershipEpoch: assignment.Epoch,
	}})
	if err != nil || len(got) != 1 || got[0].Fresh {
		t.Fatalf("ReadRuntime(expired) = %+v, %v", got, err)
	}
}

func TestRedisWorkerRuntimeRejectsEmptyReportBeyondAcknowledgedSnapshot(t *testing.T) {
	fixture := newRedisTestFixture(t)
	now := redisTestNow()
	if err := fixture.Adapter.ReplaceSession(context.Background(), workerruntime.Session{
		WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a",
		AckedSnapshotVersion: 3, AckedOwnershipEpoch: 9,
	}, time.Minute); err != nil {
		t.Fatalf("ReplaceSession(): %v", err)
	}
	for name, report := range map[string]workerruntime.Report{
		"version": {
			WorkerID: "worker-a", SessionID: "session-a", Sequence: 1,
			SnapshotVersion: 4, OwnershipEpoch: 9, ObservedAt: now,
		},
		"epoch": {
			WorkerID: "worker-a", SessionID: "session-a", Sequence: 1,
			SnapshotVersion: 3, OwnershipEpoch: 10, ObservedAt: now,
		},
	} {
		t.Run(name, func(t *testing.T) {
			if err := fixture.Adapter.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, workerruntime.ErrSnapshotMismatch) {
				t.Fatalf("ReplaceRuntime() error = %v, want ErrSnapshotMismatch", err)
			}
		})
	}
}

func TestRedisWorkerRuntimeAcceptsEmptyRead(t *testing.T) {
	fixture := newRedisTestFixture(t)
	got, err := fixture.Adapter.ReadRuntime(context.Background(), nil)
	if err != nil || got == nil || len(got) != 0 {
		t.Fatalf("ReadRuntime(empty) = %#v, %v", got, err)
	}
}