internal/store/postgres/message_concurrent_integration_test.go

Rama main · réplica de lectura del repositorio original.

package postgres

import (
	"context"
	"sync"
	"testing"
	"time"

	"telesrv/internal/domain"
)

// TestSendPrivateTextConcurrentNoPtsGap is the core correctness proof for "allocate pts inside a
// PG transaction": with N messages sent concurrently sender→recipient, after all commit
// successfully the recipient's account event pts must be strictly contiguous 1..N (no gaps, no
// duplicates, no loss), and MaxContiguousPts == N. box id may still come from an external counter,
// while pts is advanced by user_update_watermarks within the same PG tx.
func TestSendPrivateTextConcurrentNoPtsGap(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	suffix := randomSuffix(t)

	users := NewUserStore(pool)
	sender, err := users.Create(ctx, domain.User{
		AccessHash: 51,
		Phone:      "+1999" + suffix + "01",
		FirstName:  "ConcSender",
	})
	if err != nil {
		t.Fatalf("create sender: %v", err)
	}
	recipient, err := users.Create(ctx, domain.User{
		AccessHash: 52,
		Phone:      "+1999" + suffix + "02",
		FirstName:  "ConcRecipient",
	})
	if err != nil {
		t.Fatalf("create recipient: %v", err)
	}
	ids := []int64{sender.ID, recipient.ID}
	t.Cleanup(func() {
		// Clean up child tables in FK dependency order (message_boxes.from_user_id is RESTRICT) before deleting users.
		_, _ = pool.Exec(ctx, "DELETE FROM dispatch_outbox WHERE target_user_id = ANY($1::bigint[])", ids)
		_, _ = pool.Exec(ctx, "DELETE FROM user_update_events WHERE user_id = ANY($1::bigint[])", ids)
		_, _ = pool.Exec(ctx, "DELETE FROM message_boxes WHERE owner_user_id = ANY($1::bigint[])", ids)
		_, _ = pool.Exec(ctx, "DELETE FROM private_messages WHERE sender_user_id = ANY($1::bigint[])", ids)
		_, _ = pool.Exec(ctx, "DELETE FROM dialogs WHERE user_id = ANY($1::bigint[])", ids)
		_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = ANY($1::bigint[])", ids)
	})
	var initialWatermarks int
	if err := pool.QueryRow(ctx, `
SELECT count(*)::int
FROM user_update_watermarks
WHERE user_id = ANY($1::bigint[])`, ids).Scan(&initialWatermarks); err != nil {
		t.Fatalf("count initial watermarks: %v", err)
	}
	if initialWatermarks != 0 {
		t.Fatalf("initial watermarks = %d, want 0 so concurrency covers first upsert", initialWatermarks)
	}

	messages := NewMessageStore(pool, WithMessageAllocators(&perUserCounterAllocator{}))

	const n = 200
	base := time.Now().UnixNano()
	date := int(time.Now().Unix())
	errs := make([]error, n)
	recipPts := make([]int, n)
	sem := make(chan struct{}, 32)
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		sem <- struct{}{}
		go func(i int) {
			defer wg.Done()
			defer func() { <-sem }()
			res, err := messages.SendPrivateText(ctx, domain.SendPrivateTextRequest{
				SenderUserID:    sender.ID,
				RecipientUserID: recipient.ID,
				RandomID:        base + int64(i),
				Message:         "concurrent body",
				Date:            date,
			})
			errs[i] = err
			recipPts[i] = res.RecipientMessage.Pts
		}(i)
	}
	wg.Wait()

	for i, err := range errs {
		if err != nil {
			t.Fatalf("send %d: %v", i, err)
		}
	}

	// In-process check for whether the allocator assigned duplicate pts to the recipient.
	seen := map[int]int{}
	for i := 0; i < n; i++ {
		if prev, ok := seen[recipPts[i]]; ok {
			t.Errorf("recipient pts %d assigned by both send #%d and #%d (allocator concurrent duplicate)", recipPts[i], prev, i)
		}
		seen[recipPts[i]] = i
	}

	// The recipient should have n events with pts contiguous 1..n (no gaps, no duplicates, no loss).
	events := NewUpdateEventStore(pool)
	got, err := events.ListAfter(ctx, recipient.ID, 0, n+10)
	if err != nil {
		t.Fatalf("list recipient events: %v", err)
	}
	if len(got) != n {
		t.Fatalf("recipient events = %d, want %d (no loss, no duplicates)", len(got), n)
	}
	for i, ev := range got {
		if ev.Pts != i+1 {
			t.Fatalf("recipient event[%d].Pts = %d, want %d (pts must be contiguous with no gaps)", i, ev.Pts, i+1)
		}
	}
	contig, err := events.MaxContiguousPts(ctx, recipient.ID)
	if err != nil {
		t.Fatalf("MaxContiguousPts: %v", err)
	}
	if contig != n {
		t.Fatalf("MaxContiguousPts(recipient) = %d, want %d", contig, n)
	}
}