internal/store/postgres/channel_send_idempotency_integration_test.go
Rama main · réplica de lectura del repositorio original.
package postgres
import (
"bytes"
"context"
"crypto/sha256"
"errors"
"os"
"strings"
"sync"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"telesrv/deploy"
"telesrv/internal/domain"
"telesrv/internal/store"
)
func TestChannelSendFingerprintReplayPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
owner, err := users.Create(ctx, domain.User{AccessHash: 181, Phone: "+1781" + suffix + "01", FirstName: "ChannelReplayOwner"})
if err != nil {
t.Fatalf("create owner: %v", err)
}
var channelID int64
t.Cleanup(func() {
if channelID != 0 {
_, _ = pool.Exec(ctx, `DELETE FROM channels WHERE id = $1`, channelID)
}
_, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, owner.ID)
})
channels := NewChannelStore(pool)
created, err := channels.CreateChannel(ctx, domain.CreateChannelRequest{
CreatorUserID: owner.ID,
Title: "Channel replay " + suffix,
Megagroup: true,
Date: 1700100000,
})
if err != nil {
t.Fatalf("create channel: %v", err)
}
channelID = created.Channel.ID
base := domain.SendChannelMessageRequest{
UserID: owner.ID,
ChannelID: channelID,
RandomID: 781001,
Message: "immutable original",
Entities: []domain.MessageEntity{{
Type: domain.MessageEntityBold,
Offset: 0,
Length: 9,
}},
Date: 1700100001,
}
wantFingerprint, err := store.ChannelSendFingerprint(base)
if err != nil {
t.Fatalf("fingerprint base: %v", err)
}
first, err := channels.SendChannelMessage(ctx, base)
if err != nil {
t.Fatalf("first send: %v", err)
}
var storedFingerprint []byte
if err := pool.QueryRow(ctx, `SELECT request_fingerprint FROM channel_messages WHERE channel_id = $1 AND id = $2`, channelID, first.Message.ID).Scan(&storedFingerprint); err != nil {
t.Fatalf("load fingerprint: %v", err)
}
if !bytes.Equal(storedFingerprint, wantFingerprint) {
t.Fatalf("stored fingerprint = %x, want %x", storedFingerprint, wantFingerprint)
}
type durableState struct {
pts int
events int
rows int
}
loadState := func(randomID int64) durableState {
t.Helper()
var state durableState
if err := pool.QueryRow(ctx, `SELECT pts FROM channels WHERE id = $1`, channelID).Scan(&state.pts); err != nil {
t.Fatalf("load channel pts: %v", err)
}
if err := pool.QueryRow(ctx, `SELECT count(*) FROM channel_update_events WHERE channel_id = $1`, channelID).Scan(&state.events); err != nil {
t.Fatalf("count channel events: %v", err)
}
if err := pool.QueryRow(ctx, `SELECT count(*) FROM channel_messages WHERE channel_id = $1 AND sender_user_id = $2 AND random_id = $3`, channelID, owner.ID, randomID).Scan(&state.rows); err != nil {
t.Fatalf("count random receipt: %v", err)
}
return state
}
before := loadState(base.RandomID)
exact := base
exact.Date += 100 // execution time is not part of immutable intent
replay, err := channels.SendChannelMessage(ctx, exact)
if err != nil {
t.Fatalf("exact replay: %v", err)
}
if !replay.Duplicate || replay.Message.ID != first.Message.ID || replay.Event.Pts != first.Event.Pts {
t.Fatalf("exact replay = %+v, want first id=%d pts=%d", replay, first.Message.ID, first.Event.Pts)
}
if after := loadState(base.RandomID); after != before {
t.Fatalf("exact replay mutated state = %+v, want %+v", after, before)
}
conflicts := []struct {
name string
mutate func(*domain.SendChannelMessageRequest)
}{
{name: "body", mutate: func(req *domain.SendChannelMessageRequest) { req.Message = "changed body" }},
{name: "media", mutate: func(req *domain.SendChannelMessageRequest) {
req.Media = &domain.MessageMedia{Kind: domain.MessageMediaKindPhoto, Photo: &domain.Photo{ID: 781, AccessHash: 782, DCID: 2}}
}},
{name: "reply", mutate: func(req *domain.SendChannelMessageRequest) {
req.ReplyTo = &domain.MessageReply{MessageID: first.Message.ID}
}},
{name: "group", mutate: func(req *domain.SendChannelMessageRequest) { req.GroupedID = 781003 }},
}
for _, tc := range conflicts {
t.Run("conflict_"+tc.name, func(t *testing.T) {
changed := base
tc.mutate(&changed)
if _, err := channels.SendChannelMessage(ctx, changed); !errors.Is(err, domain.ErrMessageRandomIDDuplicate) {
t.Fatalf("changed %s replay err = %v, want ErrMessageRandomIDDuplicate", tc.name, err)
}
if after := loadState(base.RandomID); after != before {
t.Fatalf("changed %s replay mutated state = %+v, want %+v", tc.name, after, before)
}
})
}
if _, err := channels.EditChannelMessage(ctx, domain.EditChannelMessageRequest{
UserID: owner.ID, ChannelID: channelID, ID: first.Message.ID, Message: "edited current", EditDate: 1700100002,
}); err != nil {
t.Fatalf("edit channel message: %v", err)
}
editedReplay, err := channels.SendChannelMessage(ctx, exact)
if err != nil {
t.Fatalf("replay edited message: %v", err)
}
if !editedReplay.Duplicate || editedReplay.Message.Body != "edited current" || editedReplay.Event.Pts != first.Event.Pts {
t.Fatalf("edited replay = %+v, want current projection with first pts", editedReplay)
}
deleted, err := channels.DeleteChannelMessages(ctx, domain.DeleteChannelMessagesRequest{
UserID: owner.ID, ChannelID: channelID, IDs: []int{first.Message.ID}, Date: 1700100003,
})
if err != nil {
t.Fatalf("delete channel message: %v", err)
}
deletedReplay, err := channels.SendChannelMessage(ctx, exact)
if err != nil {
t.Fatalf("replay deleted message: %v", err)
}
if !deletedReplay.Duplicate || deletedReplay.Message.Body != base.Message || deletedReplay.Message.ID != first.Message.ID || deletedReplay.ReplayDeleteEvent == nil || deletedReplay.ReplayDeleteEvent.Pts != deleted.Event.Pts {
t.Fatalf("deleted replay = %+v, want immutable first snapshot + delete receipt %+v", deletedReplay, deleted.Event)
}
// A raw request-boundary fingerprint must be stored byte-for-byte rather
// than replaced with the domain fallback.
raw := sha256.Sum256([]byte("raw channel TL intent"))
rawReq := domain.SendChannelMessageRequest{
UserID: owner.ID, ChannelID: channelID, RandomID: 781002, Message: "raw fingerprint", Date: 1700100010,
IdempotencyFingerprint: raw[:],
}
rawSent, err := channels.SendChannelMessage(ctx, rawReq)
if err != nil {
t.Fatalf("raw fingerprint send: %v", err)
}
storedFingerprint = nil
if err := pool.QueryRow(ctx, `SELECT request_fingerprint FROM channel_messages WHERE channel_id = $1 AND id = $2`, channelID, rawSent.Message.ID).Scan(&storedFingerprint); err != nil {
t.Fatalf("load raw fingerprint: %v", err)
}
if !bytes.Equal(storedFingerprint, raw[:]) {
t.Fatalf("stored raw fingerprint = %x, want %x", storedFingerprint, raw)
}
// Simulate a rolling old writer that omits the new column. The empty
// default keeps the write compatible, but it is never accepted as replay.
legacyID := rawSent.Message.ID + 100
legacyRandomID := int64(781099)
if _, err := pool.Exec(ctx, `
INSERT INTO channel_messages (channel_id, id, random_id, sender_user_id, from_peer_id, message_date, pts, body)
VALUES ($1,$2,$3,$4,$4,$5,$6,$7)`, channelID, legacyID, legacyRandomID, owner.ID, 1700100020, rawSent.Event.Pts+100, "legacy unknown intent"); err != nil {
t.Fatalf("old-writer insert without fingerprint: %v", err)
}
var legacyFingerprint []byte
if err := pool.QueryRow(ctx, `SELECT request_fingerprint FROM channel_messages WHERE channel_id=$1 AND id=$2`, channelID, legacyID).Scan(&legacyFingerprint); err != nil {
t.Fatalf("load legacy fingerprint: %v", err)
}
if len(legacyFingerprint) != 0 {
t.Fatalf("legacy fingerprint length = %d, want empty", len(legacyFingerprint))
}
legacyReq := domain.SendChannelMessageRequest{UserID: owner.ID, ChannelID: channelID, RandomID: legacyRandomID, Message: "legacy unknown intent", Date: 1700100021}
legacyExpected, err := store.ChannelSendFingerprint(legacyReq)
if err != nil {
t.Fatalf("fingerprint legacy retry: %v", err)
}
if _, _, err := channels.LookupChannelSendReplay(ctx, domain.ChannelSendReplayRequest{
ChannelID: channelID, SenderUserID: owner.ID, RandomID: legacyRandomID, IdempotencyFingerprint: legacyExpected,
}); !errors.Is(err, domain.ErrMessageRandomIDDuplicate) {
t.Fatalf("legacy empty lookup err = %v, want ErrMessageRandomIDDuplicate", err)
}
if _, err := channels.SendChannelMessage(ctx, legacyReq); !errors.Is(err, domain.ErrMessageRandomIDDuplicate) {
t.Fatalf("legacy empty send err = %v, want ErrMessageRandomIDDuplicate", err)
}
}
func TestChannelSendFingerprintConcurrentRacePostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
owner, err := users.Create(ctx, domain.User{AccessHash: 191, Phone: "+1781" + suffix + "11", FirstName: "ChannelRaceOwner"})
if err != nil {
t.Fatalf("create owner: %v", err)
}
var channelIDs []int64
t.Cleanup(func() {
if len(channelIDs) != 0 {
_, _ = pool.Exec(ctx, `DELETE FROM channels WHERE id = ANY($1::bigint[])`, channelIDs)
}
_, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, owner.ID)
})
newChannel := func(title string) int64 {
t.Helper()
created, err := NewChannelStore(pool).CreateChannel(ctx, domain.CreateChannelRequest{CreatorUserID: owner.ID, Title: title + suffix, Megagroup: true, Date: 1700110000})
if err != nil {
t.Fatalf("create %s channel: %v", title, err)
}
channelIDs = append(channelIDs, created.Channel.ID)
return created.Channel.ID
}
run := func(reqs [2]domain.SendChannelMessageRequest) ([2]domain.SendChannelMessageResult, [2]error) {
t.Helper()
var results [2]domain.SendChannelMessageResult
var errs [2]error
start := make(chan struct{})
var wg sync.WaitGroup
for i := range reqs {
wg.Add(1)
go func(i int) {
defer wg.Done()
<-start
results[i], errs[i] = NewChannelStore(pool).SendChannelMessage(ctx, reqs[i])
}(i)
}
close(start)
wg.Wait()
return results, errs
}
exactChannelID := newChannel("exact race ")
exactReq := domain.SendChannelMessageRequest{
UserID: owner.ID, ChannelID: exactChannelID, RandomID: 791001, Message: "same intent", Date: 1700110001,
IdempotencyPreflighted: true,
}
exactResults, exactErrs := run([2]domain.SendChannelMessageRequest{exactReq, exactReq})
for i, err := range exactErrs {
if err != nil {
t.Fatalf("exact race result[%d] err = %v", i, err)
}
}
if exactResults[0].Message.ID != exactResults[1].Message.ID || exactResults[0].Duplicate == exactResults[1].Duplicate {
t.Fatalf("exact race results = %+v / %+v, want same id and one duplicate", exactResults[0], exactResults[1])
}
assertChannelRandomReceiptCount(t, ctx, pool, exactChannelID, owner.ID, exactReq.RandomID, 1)
conflictChannelID := newChannel("conflict race ")
conflictA := domain.SendChannelMessageRequest{
UserID: owner.ID, ChannelID: conflictChannelID, RandomID: 791002, Message: "intent A", Date: 1700110010,
IdempotencyPreflighted: true,
}
conflictB := conflictA
conflictB.Message = "intent B"
conflictResults, conflictErrs := run([2]domain.SendChannelMessageRequest{conflictA, conflictB})
nilCount, duplicateErrCount := 0, 0
for _, err := range conflictErrs {
switch {
case err == nil:
nilCount++
case errors.Is(err, domain.ErrMessageRandomIDDuplicate):
duplicateErrCount++
default:
t.Fatalf("conflicting race unexpected err = %v; results=%+v", err, conflictResults)
}
}
if nilCount != 1 || duplicateErrCount != 1 {
t.Fatalf("conflicting race errors = %v, want one success and one duplicate", conflictErrs)
}
assertChannelRandomReceiptCount(t, ctx, pool, conflictChannelID, owner.ID, conflictA.RandomID, 1)
}
func TestChannelSendFingerprintSingleConnectionConflictLookupPostgres(t *testing.T) {
dsn := os.Getenv("TELESRV_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set TELESRV_TEST_POSTGRES_DSN to run postgres integration test")
}
setupPool := testPool(t)
setupCtx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(setupPool)
owner, err := users.Create(setupCtx, domain.User{AccessHash: 192, Phone: "+1781" + suffix + "21", FirstName: "OneConnectionOwner"})
if err != nil {
t.Fatalf("create owner: %v", err)
}
var channelIDs []int64
t.Cleanup(func() {
cleanupCtx := context.Background()
if len(channelIDs) != 0 {
_, _ = setupPool.Exec(cleanupCtx, `DELETE FROM channels WHERE id = ANY($1::bigint[])`, channelIDs)
}
_, _ = setupPool.Exec(cleanupCtx, `DELETE FROM users WHERE id = $1`, owner.ID)
})
setupChannels := NewChannelStore(setupPool)
created, err := setupChannels.CreateChannel(setupCtx, domain.CreateChannelRequest{
CreatorUserID: owner.ID, Title: "one connection " + suffix, Megagroup: true, Date: 1700120000,
})
if err != nil {
t.Fatalf("create ordinary channel: %v", err)
}
channelIDs = append(channelIDs, created.Channel.ID)
ordinaryReq := domain.SendChannelMessageRequest{
UserID: owner.ID, ChannelID: created.Channel.ID, RandomID: 792001, Message: "single pool exact", Date: 1700120001,
IdempotencyPreflighted: true,
}
broadcast, err := setupChannels.CreateChannel(setupCtx, domain.CreateChannelRequest{
CreatorUserID: owner.ID, Title: "one connection mono " + suffix, Broadcast: true, Date: 1700120010,
})
if err != nil {
t.Fatalf("create broadcast: %v", err)
}
channelIDs = append(channelIDs, broadcast.Channel.ID)
enabled, err := setupChannels.SetPaidMessagesPrice(setupCtx, owner.ID, broadcast.Channel.ID, 0, true)
if err != nil {
t.Fatalf("enable monoforum: %v", err)
}
monoID := enabled.Channel.LinkedMonoforumID
channelIDs = append(channelIDs, monoID)
// Fixture creation itself has legacy allocator paths that require more than
// one connection. Constrain only the send/replay path under test.
cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
t.Fatalf("parse postgres config: %v", err)
}
cfg.MaxConns = 1
cfg.MinConns = 0
pool, err := pgxpool.NewWithConfig(context.Background(), cfg)
if err != nil {
t.Fatalf("open one-connection pool: %v", err)
}
t.Cleanup(pool.Close)
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
t.Cleanup(cancel)
msgIDs := &singleConnectionMessageIDAllocator{current: make(map[int64]int)}
for _, id := range []int64{created.Channel.ID, monoID} {
var current int
if err := setupPool.QueryRow(setupCtx, `SELECT COALESCE(MAX(id), 0) FROM channel_messages WHERE channel_id=$1`, id).Scan(¤t); err != nil {
t.Fatalf("seed message allocator for channel %d: %v", id, err)
}
msgIDs.current[id] = current
}
oneConnectionStore := func() *ChannelStore {
return NewChannelStore(pool, WithChannelAllocators(nil, msgIDs))
}
var ordinaryResults [2]domain.SendChannelMessageResult
var ordinaryErrs [2]error
start := make(chan struct{})
var wg sync.WaitGroup
for i := range ordinaryResults {
wg.Add(1)
go func(i int) {
defer wg.Done()
<-start
ordinaryResults[i], ordinaryErrs[i] = oneConnectionStore().SendChannelMessage(ctx, ordinaryReq)
}(i)
}
close(start)
wg.Wait()
for i, err := range ordinaryErrs {
if err != nil {
t.Fatalf("one-connection ordinary result[%d] err = %v", i, err)
}
}
if ordinaryResults[0].Message.ID != ordinaryResults[1].Message.ID || ordinaryResults[0].Duplicate == ordinaryResults[1].Duplicate {
t.Fatalf("one-connection ordinary results = %+v / %+v, want same id and one duplicate", ordinaryResults[0], ordinaryResults[1])
}
monoReq := domain.SendMonoforumMessageRequest{
MonoforumID: monoID, SenderUserID: owner.ID,
SavedPeer: domain.Peer{Type: domain.PeerTypeUser, ID: owner.ID},
RandomID: 792002, Message: "single pool mono exact", Date: 1700120011,
IdempotencyPreflighted: true,
}
var monoResults [2]domain.SendChannelMessageResult
var monoErrs [2]error
start = make(chan struct{})
for i := range monoResults {
wg.Add(1)
go func(i int) {
defer wg.Done()
<-start
monoResults[i], monoErrs[i] = oneConnectionStore().SendMonoforumMessage(ctx, monoReq)
}(i)
}
close(start)
wg.Wait()
for i, err := range monoErrs {
if err != nil {
t.Fatalf("one-connection monoforum result[%d] err = %v", i, err)
}
}
if monoResults[0].Message.ID != monoResults[1].Message.ID || monoResults[0].Duplicate == monoResults[1].Duplicate {
t.Fatalf("one-connection monoforum results = %+v / %+v, want same id and one duplicate", monoResults[0], monoResults[1])
}
}
type singleConnectionMessageIDAllocator struct {
mu sync.Mutex
current map[int64]int
}
func (a *singleConnectionMessageIDAllocator) NextChannelMessageID(_ context.Context, channelID int64) (int, error) {
a.mu.Lock()
defer a.mu.Unlock()
a.current[channelID]++
return a.current[channelID], nil
}
func (a *singleConnectionMessageIDAllocator) CurrentChannelMessageID(_ context.Context, channelID int64) (int, error) {
a.mu.Lock()
defer a.mu.Unlock()
return a.current[channelID], nil
}
func assertChannelRandomReceiptCount(t *testing.T, ctx context.Context, pool *pgxpool.Pool, channelID, senderUserID, randomID int64, want int) {
t.Helper()
var got int
if err := pool.QueryRow(ctx, `SELECT count(*) FROM channel_messages WHERE channel_id=$1 AND sender_user_id=$2 AND random_id=$3`, channelID, senderUserID, randomID).Scan(&got); err != nil {
t.Fatalf("count channel random receipt: %v", err)
}
if got != want {
t.Fatalf("channel random receipt count = %d, want %d", got, want)
}
}
func TestChannelSendFingerprintMigrationRoundTripPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
downSQL, err := deploy.Migrations.ReadFile("migrations/0078_channel_send_fingerprint.down.sql")
if err != nil {
t.Fatalf("read 0078 down: %v", err)
}
upSQL, err := deploy.Migrations.ReadFile("migrations/0078_channel_send_fingerprint.up.sql")
if err != nil {
t.Fatalf("read 0078 up: %v", err)
}
tx, err := pool.Begin(ctx)
if err != nil {
t.Fatalf("begin 0078 round trip: %v", err)
}
defer func() { _ = tx.Rollback(context.Background()) }()
if _, err := tx.Exec(ctx, string(downSQL)); err != nil {
t.Fatalf("0078 down: %v", err)
}
if _, err := tx.Exec(ctx, string(upSQL)); err != nil {
t.Fatalf("0078 up: %v", err)
}
var defaultExpr string
var constraintExists bool
if err := tx.QueryRow(ctx, `
SELECT column_default,
EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'channel_messages_request_fingerprint_size')
FROM information_schema.columns
WHERE table_schema='public' AND table_name='channel_messages' AND column_name='request_fingerprint'`).Scan(&defaultExpr, &constraintExists); err != nil {
t.Fatalf("inspect 0078: %v", err)
}
if !strings.Contains(defaultExpr, `\x`) || !constraintExists {
t.Fatalf("0078 default=%q constraint=%v, want empty bytea rolling default + size constraint", defaultExpr, constraintExists)
}
}