feat: add reusable MySQL outbox and HTTP lifecycle example
This commit is contained in:
226
adapters/outbox/mysql/commit_loss_integration_test.go
Normal file
226
adapters/outbox/mysql/commit_loss_integration_test.go
Normal file
@@ -0,0 +1,226 @@
|
||||
package outbox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
// commitLossGateway 只用于无TLS的隔离MySQL回归；捕获COMMIT的真实成功包后丢弃并断连。
|
||||
// 不伪造SQL结果，也不修改数据库中的提交状态。
|
||||
type commitLossGateway struct {
|
||||
listener net.Listener
|
||||
dropped chan struct{}
|
||||
armed atomic.Bool
|
||||
mu sync.Mutex
|
||||
closing bool
|
||||
connections map[net.Conn]struct{}
|
||||
wg sync.WaitGroup
|
||||
accepted chan struct{}
|
||||
}
|
||||
|
||||
func newCommitLossGateway(t *testing.T, target string) *commitLossGateway {
|
||||
t.Helper()
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
g := &commitLossGateway{listener: listener, dropped: make(chan struct{}, 1), connections: map[net.Conn]struct{}{}, accepted: make(chan struct{})}
|
||||
go func() {
|
||||
defer close(g.accepted)
|
||||
for {
|
||||
down, err := listener.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
g.wg.Add(1)
|
||||
go func() {
|
||||
defer g.wg.Done()
|
||||
defer down.Close()
|
||||
up, err := net.DialTimeout("tcp", target, 3*time.Second)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer up.Close()
|
||||
g.mu.Lock()
|
||||
if g.closing {
|
||||
g.mu.Unlock()
|
||||
return
|
||||
}
|
||||
g.connections[down] = struct{}{}
|
||||
g.connections[up] = struct{}{}
|
||||
g.mu.Unlock()
|
||||
defer func() { g.mu.Lock(); delete(g.connections, down); delete(g.connections, up); g.mu.Unlock() }()
|
||||
var committing atomic.Bool
|
||||
done := make(chan struct{}, 2)
|
||||
go func() {
|
||||
defer func() { done <- struct{}{} }()
|
||||
for {
|
||||
packet, err := readMySQLPacket(down)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if packet[3] == 0 && len(packet) > 4 && packet[4] == 3 {
|
||||
committing.Store(strings.EqualFold(strings.TrimSpace(string(packet[5:])), "COMMIT"))
|
||||
}
|
||||
if writeMySQLPacket(up, packet) != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
defer func() { done <- struct{}{} }()
|
||||
for {
|
||||
packet, err := readMySQLPacket(up)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if packet[3] == 1 && committing.Swap(false) && len(packet) > 4 && packet[4] == 0 && g.armed.CompareAndSwap(true, false) {
|
||||
select {
|
||||
case g.dropped <- struct{}{}:
|
||||
default:
|
||||
t.Error("unconsumed COMMIT fault marker")
|
||||
}
|
||||
return
|
||||
}
|
||||
if writeMySQLPacket(down, packet) != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
<-done
|
||||
down.Close()
|
||||
up.Close()
|
||||
<-done
|
||||
}()
|
||||
}
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
listener.Close()
|
||||
<-g.accepted
|
||||
g.mu.Lock()
|
||||
g.closing = true
|
||||
for c := range g.connections {
|
||||
c.Close()
|
||||
}
|
||||
g.mu.Unlock()
|
||||
g.wg.Wait()
|
||||
})
|
||||
return g
|
||||
}
|
||||
func readMySQLPacket(r io.Reader) ([]byte, error) {
|
||||
var header [4]byte
|
||||
if _, err := io.ReadFull(r, header[:]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
size := int(binary.LittleEndian.Uint32(header[:]) & 0x00ffffff)
|
||||
packet := make([]byte, 4+size)
|
||||
copy(packet, header[:])
|
||||
_, err := io.ReadFull(r, packet[4:])
|
||||
return packet, err
|
||||
}
|
||||
func writeMySQLPacket(w io.Writer, b []byte) error {
|
||||
for len(b) > 0 {
|
||||
n, err := w.Write(b)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n == 0 {
|
||||
return io.ErrShortWrite
|
||||
}
|
||||
b = b[n:]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestIntegrationSenderCommitResponseLoss(t *testing.T) {
|
||||
direct, dsn := integrationDB(t)
|
||||
cfg, err := mysql.ParseDSN(dsn)
|
||||
if err != nil || cfg.TLSConfig != "" {
|
||||
t.Fatal("fault requires isolated plaintext MySQL")
|
||||
}
|
||||
gateway := newCommitLossGateway(t, cfg.Addr)
|
||||
cfg.Addr = gateway.listener.Addr().String()
|
||||
ctx, stop := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer stop()
|
||||
proxied, err := Open(ctx, cfg.FormatDSN())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer proxied.Close()
|
||||
for _, phase := range []string{"enqueue", "claim", "confirm"} {
|
||||
t.Run(phase, func(t *testing.T) {
|
||||
s := testStore(t, proxied)
|
||||
fresh, _ := businessStore(direct, s.namespace, s.queue)
|
||||
id := testID(t)
|
||||
in := testRequest(s.queue)
|
||||
if phase == "enqueue" {
|
||||
gateway.armed.Store(true)
|
||||
if _, err = s.Enqueue(ctx, id, "objects/test", in); !errors.Is(err, ErrCommitUnconfirmed) {
|
||||
t.Fatal("enqueue reported confirmed", err)
|
||||
}
|
||||
dup, e := fresh.Enqueue(ctx, id, "objects/test", in)
|
||||
if e != nil || !dup || countRows(t, fresh, "sender_jobs") != 1 {
|
||||
t.Fatal("local commit not reconciled", e)
|
||||
}
|
||||
} else {
|
||||
if _, err = s.Enqueue(ctx, id, "objects/test", in); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if phase == "claim" {
|
||||
gateway.armed.Store(true)
|
||||
}
|
||||
claim, e := s.claim(ctx)
|
||||
if phase == "claim" {
|
||||
if !errors.Is(e, ErrCommitUnconfirmed) {
|
||||
t.Fatal("claim reported confirmed", e)
|
||||
}
|
||||
r := status(t, fresh, id)
|
||||
if r.State != "SENDING" || r.SendCount != 1 {
|
||||
t.Fatal("claim was not actually committed")
|
||||
}
|
||||
execSQL(t, direct, `UPDATE sender_outbox SET wake_at=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND operation_id=?`, s.namespace, id)
|
||||
r, e = fresh.claim(ctx)
|
||||
if e != nil || r.SendCount != 2 || r.IdempotencyKey != "outbox-task-"+id {
|
||||
t.Fatal("unknown claim did not recover", e)
|
||||
}
|
||||
} else {
|
||||
if e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
receipt := fakeReceipt(t, s)
|
||||
gateway.armed.Store(true)
|
||||
if e = s.finish(ctx, claim, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(e, ErrCommitUnconfirmed) {
|
||||
t.Fatal("confirmation reported confirmed", e)
|
||||
}
|
||||
r := status(t, fresh, id)
|
||||
if r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != receipt.Task.ID {
|
||||
t.Fatal("confirmed commit not visible")
|
||||
}
|
||||
if _, e = fresh.claim(ctx); !errors.Is(e, ErrIdle) {
|
||||
t.Fatal("confirmed item eligible for resend", e)
|
||||
}
|
||||
var task string
|
||||
if e = direct.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=?`, s.namespace, id).Scan(&task); e != nil || task != receipt.Task.ID {
|
||||
t.Fatal("business confirmation not atomic", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-gateway.dropped:
|
||||
case <-ctx.Done():
|
||||
t.Fatal("actual COMMIT OK not intercepted")
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Log("real COMMIT OK discarded for business enqueue, sender claim and local acceptance; fresh stores reconciled persisted identity and state")
|
||||
}
|
||||
Reference in New Issue
Block a user