test(P1.2): примитивы индексации — tombstone O(1), компакция, receipt/uuid-индексы, Reset

app/models/message_index_test.go

@@ -0,0 +1,163 @@
+// app/models/message_index_test.go
+// Тесты индексации сообщений (P1.2): tombstone-удаление, компакция, индексы
+// receipt/uuid. Проверяют ключевые свойства, на которых держится O(1)-удаление
+// и O(in-flight) visibility-перерасчёт.
+package models
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+// newTestQueue — очередь с последовательно добавленными сообщениями.
+func newTestQueue(uuids ...string) *Queue {
+	q := &Queue{Name: "t", Duplicates: map[string]time.Time{}}
+	for _, u := range uuids {
+		q.AppendMessage(SqsMessage{Uuid: u, MessageBody: u})
+	}
+	return q
+}
+
+// TestQueue_RemoveIsTombstoneO1 — удаление не двигает срез и снимает индексы.
+func TestQueue_RemoveIsTombstoneO1(t *testing.T) {
+	q := newTestQueue("a", "b", "c")
+	if q.LiveCount() != 3 {
+		t.Fatalf("LiveCount = %d, want 3", q.LiveCount())
+	}
+	if i, ok := q.IndexOfUUID("b"); !ok || i != 1 {
+		t.Fatalf("IndexOfUUID(b) = %d,%v; want 1,true", i, ok)
+	}
+
+	q.RemoveIndex(1)
+
+	if q.LiveCount() != 2 {
+		t.Fatalf("LiveCount после удаления = %d, want 2", q.LiveCount())
+	}
+	if !q.IsDead(1) {
+		t.Fatal("слот 1 должен быть tombstone")
+	}
+	if len(q.Messages) != 3 {
+		t.Fatalf("физический размер среза изменился (%d) — удаление должно быть tombstone O(1)", len(q.Messages))
+	}
+	if _, ok := q.IndexOfUUID("b"); ok {
+		t.Fatal("удалённый Uuid не должен находиться в индексе")
+	}
+	// Оставшиеся индексы не должны «съехать».
+	if i, ok := q.IndexOfUUID("c"); !ok || i != 2 {
+		t.Fatalf("IndexOfUUID(c) = %d,%v; want 2,true", i, ok)
+	}
+}
+
+// TestQueue_CompactKeepsOrder — компакция сохраняет порядок живых и перестраивает индексы.
+func TestQueue_CompactKeepsOrder(t *testing.T) {
+	q := newTestQueue("a", "b", "c", "d")
+	// Помечаем сообщение "c" in-flight, чтобы проверить сохранение receipt-индекса.
+	q.Messages[2].ReceiptHandle = "c#1"
+	q.MarkInFlight(2, "c#1")
+
+	q.RemoveIndex(1) // b
+	q.RemoveIndex(3) // d
+	q.Compact()
+
+	if q.LiveCount() != 2 || len(q.Messages) != 2 {
+		t.Fatalf("после компакции LiveCount=%d len=%d, want 2/2", q.LiveCount(), len(q.Messages))
+	}
+	if q.deadCount != 0 {
+		t.Fatalf("deadCount после компакции = %d, want 0", q.deadCount)
+	}
+	if q.Messages[0].Uuid != "a" || q.Messages[1].Uuid != "c" {
+		t.Fatalf("порядок нарушен: %q, %q", q.Messages[0].Uuid, q.Messages[1].Uuid)
+	}
+	if i, ok := q.IndexOfUUID("c"); !ok || i != 1 {
+		t.Fatalf("IndexOfUUID(c) после компакции = %d,%v; want 1,true", i, ok)
+	}
+	if i, ok := q.IndexOfReceipt("c#1"); !ok || i != 1 {
+		t.Fatalf("IndexOfReceipt(c#1) после компакции = %d,%v; want 1,true", i, ok)
+	}
+}
+
+// TestQueue_ReceiptIndexLifecycle — регистрация/снятие in-flight.
+func TestQueue_ReceiptIndexLifecycle(t *testing.T) {
+	q := newTestQueue("a")
+	q.Messages[0].ReceiptHandle = "a#1"
+	q.MarkInFlight(0, "a#1")
+
+	if i, ok := q.IndexOfReceipt("a#1"); !ok || i != 0 {
+		t.Fatalf("IndexOfReceipt = %d,%v; want 0,true", i, ok)
+	}
+	if got := len(q.InFlightIndexes()); got != 1 {
+		t.Fatalf("InFlightIndexes len = %d, want 1", got)
+	}
+
+	q.ClearInFlight(0, "a#1")
+	if _, ok := q.IndexOfReceipt("a#1"); ok {
+		t.Fatal("снятый handle не должен находиться в индексе")
+	}
+	if got := len(q.InFlightIndexes()); got != 0 {
+		t.Fatalf("InFlightIndexes len = %d, want 0", got)
+	}
+}
+
+// TestQueue_RebuildIndexesAfterDirectReplace — защита от прямой замены среза
+// (загрузка из Redis / тесты): индексы восстанавливаются по содержимому.
+func TestQueue_RebuildIndexesAfterDirectReplace(t *testing.T) {
+	q := newTestQueue("a")
+	q.Messages = []SqsMessage{{Uuid: "x"}, {Uuid: "y", ReceiptHandle: "y#1"}}
+	q.RebuildIndexes()
+
+	if q.LiveCount() != 2 {
+		t.Fatalf("LiveCount = %d, want 2", q.LiveCount())
+	}
+	if i, ok := q.IndexOfUUID("y"); !ok || i != 1 {
+		t.Fatalf("IndexOfUUID(y) = %d,%v; want 1,true", i, ok)
+	}
+	if i, ok := q.IndexOfReceipt("y#1"); !ok || i != 1 {
+		t.Fatalf("IndexOfReceipt(y#1) = %d,%v; want 1,true", i, ok)
+	}
+}
+
+// TestQueue_CompactIfNeededThreshold — пороговая компакция чистит tombstone
+// и сохраняет число живых.
+func TestQueue_CompactIfNeededThreshold(t *testing.T) {
+	q := &Queue{Name: "t", Duplicates: map[string]time.Time{}}
+	const total = 5000
+	for i := 0; i < total; i++ {
+		q.AppendMessage(SqsMessage{Uuid: fmt.Sprintf("u%d", i)})
+	}
+	for i := 0; i < 2000; i++ {
+		q.RemoveIndex(i)
+	}
+	if q.deadCount != 2000 {
+		t.Fatalf("deadCount = %d, want 2000", q.deadCount)
+	}
+	q.CompactIfNeeded()
+	if q.deadCount != 0 || len(q.Messages) != total-2000 {
+		t.Fatalf("после CompactIfNeeded deadCount=%d len=%d, want 0/%d", q.deadCount, len(q.Messages), total-2000)
+	}
+	if q.LiveCount() != total-2000 {
+		t.Fatalf("LiveCount = %d, want %d", q.LiveCount(), total-2000)
+	}
+	// Порядок живых сохранён: первый живой — u2000.
+	if q.Messages[0].Uuid != "u2000" {
+		t.Fatalf("первый живой = %q, want u2000", q.Messages[0].Uuid)
+	}
+}
+
+// TestQueue_Reset — purge полностью очищает сообщения и индексы.
+func TestQueue_Reset(t *testing.T) {
+	q := newTestQueue("a", "b")
+	q.RemoveIndex(0)
+	q.Reset()
+
+	if q.LiveCount() != 0 || len(q.Messages) != 0 || q.deadCount != 0 {
+		t.Fatalf("после Reset: LiveCount=%d len=%d deadCount=%d", q.LiveCount(), len(q.Messages), q.deadCount)
+	}
+	if len(q.InFlightIndexes()) != 0 {
+		t.Fatal("после Reset in-flight индекс должен быть пуст")
+	}
+	// Очередь остаётся работоспособной.
+	if idx := q.AppendMessage(SqsMessage{Uuid: "c"}); idx != 0 || q.LiveCount() != 1 {
+		t.Fatalf("после Reset append: idx=%d LiveCount=%d", idx, q.LiveCount())
+	}
+}