490 lines
18 KiB
Go
490 lines
18 KiB
Go
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)
|
|
}
|
|
}
|