745 lines
23 KiB
Go
745 lines
23 KiB
Go
package message
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
|
"git.asio.asia/nixevol/NixMsg/internal/auth"
|
|
"git.asio.asia/nixevol/NixMsg/internal/protocol"
|
|
"git.asio.asia/nixevol/NixMsg/internal/store"
|
|
)
|
|
|
|
type deliveryEnv struct {
|
|
t *testing.T
|
|
app *App
|
|
db *store.DB
|
|
conns *MemoryConns
|
|
down *RecordingDownlink
|
|
nowMs int64
|
|
}
|
|
|
|
func openDeliveryEnv(t *testing.T, mutate func(*Limits)) *deliveryEnv {
|
|
t.Helper()
|
|
dir := t.TempDir()
|
|
db, err := store.Open(filepath.Join(dir, "data"), "FULL")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = db.Close() })
|
|
lim := defaultTestLimits()
|
|
lim.GraceSeconds = 60
|
|
lim.AckTimeoutSeconds = 300
|
|
lim.DeliveryWindow = 32
|
|
if mutate != nil {
|
|
mutate(&lim)
|
|
}
|
|
nowMs := int64(1_700_000_000_000)
|
|
conns := NewMemoryConns()
|
|
down := &RecordingDownlink{}
|
|
app := New(db, lim, auth.NewStubHashPool(),
|
|
WithNow(func() time.Time { return time.UnixMilli(nowMs) }),
|
|
WithLocks(auth.NewStubLoginLocks()),
|
|
WithConnRegistry(conns),
|
|
WithDownlink(down),
|
|
)
|
|
return &deliveryEnv{t: t, app: app, db: db, conns: conns, down: down, nowMs: nowMs}
|
|
}
|
|
|
|
func (e *deliveryEnv) setNow(ms int64) {
|
|
e.nowMs = ms
|
|
e.app.nowFn = func() time.Time { return time.UnixMilli(e.nowMs) }
|
|
}
|
|
|
|
func (e *deliveryEnv) online(id string, connID port.ConnID) {
|
|
e.conns.Set(id, LiveConn{ConnID: connID, MaxReceiveBytes: 0, MaxPacketSize: 0})
|
|
}
|
|
|
|
func (e *deliveryEnv) deliveryState(seq int64, endpointID string) (state, reason string) {
|
|
e.t.Helper()
|
|
err := e.db.Read.QueryRow(`SELECT state, reason FROM deliveries WHERE seq=? AND endpoint_id=?`, seq, endpointID).Scan(&state, &reason)
|
|
if err != nil {
|
|
e.t.Fatal(err)
|
|
}
|
|
return
|
|
}
|
|
|
|
func (e *deliveryEnv) msgState(sender, id string) (state, reason string) {
|
|
e.t.Helper()
|
|
err := e.db.Read.QueryRow(`SELECT state, reason FROM messages WHERE sender_id=? AND id=?`, sender, id).Scan(&state, &reason)
|
|
if err != nil {
|
|
e.t.Fatal(err)
|
|
}
|
|
return
|
|
}
|
|
|
|
func (e *deliveryEnv) seqOf(sender, id string) int64 {
|
|
e.t.Helper()
|
|
var seq int64
|
|
if err := e.db.Read.QueryRow(`SELECT seq FROM messages WHERE sender_id=? AND id=?`, sender, id).Scan(&seq); err != nil {
|
|
e.t.Fatal(err)
|
|
}
|
|
return seq
|
|
}
|
|
|
|
func keepTrue() *protocol.OfflineOpts {
|
|
ttl := int64(3600)
|
|
return &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl}
|
|
}
|
|
|
|
func TestDeliveryStateMachine(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
t.Run("F10_grace_within_keeps_pending", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
// offline_since = now → 宽限内
|
|
ctx := context.Background()
|
|
res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("g1", "bob"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if res.State != StateDispatched {
|
|
t.Fatalf("state=%s", res.State)
|
|
}
|
|
seq := e.seqOf("alice", "g1")
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryPending || reason != "" {
|
|
t.Fatalf("got %s/%s", st, reason)
|
|
}
|
|
var exp sql.NullInt64
|
|
_ = e.db.Read.QueryRow(`SELECT expire_at FROM deliveries WHERE seq=?`, seq).Scan(&exp)
|
|
if !exp.Valid || exp.Int64 != e.nowMs+60_000 {
|
|
t.Fatalf("expire_at=%v want %d", exp, e.nowMs+60_000)
|
|
}
|
|
})
|
|
|
|
t.Run("F10_grace_exceeded_dropped_offline", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
_ = e.db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`UPDATE endpoints SET offline_since = ? WHERE id=?`, e.nowMs-120_000, "bob")
|
|
return err
|
|
})
|
|
ctx := context.Background()
|
|
res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("g2", "bob"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if res.State != StateCompleted {
|
|
t.Fatalf("state=%s", res.State)
|
|
}
|
|
seq := e.seqOf("alice", "g2")
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryDropped || reason != ReasonOffline {
|
|
t.Fatalf("got %s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("F09_keep_ttl_expire_via_cleanup", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, func(l *Limits) { l.GraceSeconds = 60 })
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
req := baseSend("k1", "bob")
|
|
req.Offline = keepTrue()
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "k1")
|
|
st, _ := e.deliveryState(seq, "bob")
|
|
if st != DeliveryPending {
|
|
t.Fatalf("state=%s", st)
|
|
}
|
|
e.setNow(e.nowMs + 3600_000 + 1)
|
|
if err := e.app.CleanupOnce(ctx, e.nowMs); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryExpired || reason != ReasonTTL {
|
|
t.Fatalf("got %s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("F08_ack_timeout_not_keep_dropped", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, func(l *Limits) { l.AckTimeoutSeconds = 10 })
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("a1", "bob")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if e.down.FilterType(protocol.TypeMsg) < 1 {
|
|
t.Fatal("expected msg publish")
|
|
}
|
|
seq := e.seqOf("alice", "a1")
|
|
var pushed sql.NullString
|
|
_ = e.db.Read.QueryRow(`SELECT pushed_conn FROM deliveries WHERE seq=?`, seq).Scan(&pushed)
|
|
if !pushed.Valid {
|
|
t.Fatal("expected pushed")
|
|
}
|
|
e.setNow(e.nowMs + 11_000)
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryDropped || reason != ReasonNotAcked {
|
|
t.Fatalf("got %s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("F08_ack_timeout_keep_repush_then_expire", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, func(l *Limits) {
|
|
l.AckTimeoutSeconds = 10
|
|
})
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
req := baseSend("a2", "bob")
|
|
ttl := int64(30)
|
|
req.Offline = &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl}
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "a2")
|
|
// 未过 expire：清标记重推
|
|
e.setNow(e.nowMs + 11_000)
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, _ := e.deliveryState(seq, "bob")
|
|
if st != DeliveryPending {
|
|
t.Fatalf("want pending got %s", st)
|
|
}
|
|
var pushed sql.NullString
|
|
_ = e.db.Read.QueryRow(`SELECT pushed_conn FROM deliveries WHERE seq=?`, seq).Scan(&pushed)
|
|
// 可能已重推或仍清空后待推
|
|
// 过 expire
|
|
e.setNow(e.nowMs + 30_000)
|
|
_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`UPDATE deliveries SET pushed_conn=?, pushed_at=? WHERE seq=?`, "c-bob", e.nowMs-11_000, seq)
|
|
return err
|
|
})
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryExpired || reason != ReasonTTL {
|
|
t.Fatalf("got %s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("F12_scheduled_recall", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
delay := int64(10_000)
|
|
req := baseSend("r1", "bob")
|
|
req.DelayMs = &delay
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
data, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "r1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if data.Result != "recalled" || data.Recalled != 0 {
|
|
t.Fatalf("%+v", data)
|
|
}
|
|
st, reason := e.msgState("alice", "r1")
|
|
if st != StateCompleted || reason != ReasonRecalled {
|
|
t.Fatalf("%s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("F13_recall_race_partial", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
insertEndpoint(t, e.db, "carol", "", 1, 0)
|
|
ctx := context.Background()
|
|
_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES('g1','g','alice',?)`, e.nowMs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, m := range []string{"alice", "bob", "carol"} {
|
|
if _, err := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES('g1',?,?)`, m, e.nowMs); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
req := &protocol.Send{
|
|
V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "grp1",
|
|
To: protocol.Target{Kind: protocol.TargetGroup, ID: "g1"},
|
|
Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"},
|
|
}
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "grp1")
|
|
// bob 先确认
|
|
if _, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "grp1"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
data, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "2", ID: "grp1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if data.Result != "partial" || data.Accepted != 1 || data.Recalled != 1 {
|
|
t.Fatalf("%+v", data)
|
|
}
|
|
st, _ := e.deliveryState(seq, "carol")
|
|
if st != DeliveryRecalled {
|
|
t.Fatalf("carol=%s", st)
|
|
}
|
|
st, _ = e.deliveryState(seq, "bob")
|
|
if st != DeliveryAccepted {
|
|
t.Fatalf("bob=%s", st)
|
|
}
|
|
})
|
|
|
|
t.Run("F13_recall_pending_unpushed", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
ctx := context.Background()
|
|
req := baseSend("r2", "bob")
|
|
req.Offline = keepTrue()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
data, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "r2"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if data.Result != "recalled" || data.Recalled != 1 {
|
|
t.Fatalf("%+v", data)
|
|
}
|
|
})
|
|
|
|
t.Run("F14_receipt_when_sender_offline", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("rc1", "bob")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "rc1"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var n int
|
|
if err := e.db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE sender_id=? AND msg_id=? AND state=?`, "alice", "rc1", DeliveryAccepted).Scan(&n); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n != 1 {
|
|
t.Fatalf("receipts=%d", n)
|
|
}
|
|
// alice 上线后能推到回执
|
|
e.online("alice", "c-alice")
|
|
if err := e.app.PushPending(ctx, "alice", "c-alice"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if e.down.FilterType(protocol.TypeReceipt) < 1 {
|
|
t.Fatal("expected receipt push")
|
|
}
|
|
})
|
|
|
|
t.Run("F18_body_gone_record_zero_idempotent_kept", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, func(l *Limits) { l.RecordRetentionDays = 0 })
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
req := baseSend("f18", "bob")
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "f18"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var bodies int
|
|
_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM message_bodies`).Scan(&bodies)
|
|
if bodies != 0 {
|
|
t.Fatalf("bodies=%d", bodies)
|
|
}
|
|
var msgs int
|
|
_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE id=?`, "f18").Scan(&msgs)
|
|
if msgs != 0 {
|
|
t.Fatalf("messages=%d want 0", msgs)
|
|
}
|
|
var keys int
|
|
_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM send_keys WHERE msg_id=?`, "f18").Scan(&keys)
|
|
if keys != 1 {
|
|
t.Fatalf("send_keys=%d", keys)
|
|
}
|
|
// 防重：消息已删 → not_found
|
|
_, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req)
|
|
if protoCode(err) != protocol.CodeNotFound {
|
|
t.Fatalf("want not_found got %v", err)
|
|
}
|
|
})
|
|
|
|
t.Run("F11_dispatch_due_after_downtime", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
delay := int64(60_000)
|
|
req := baseSend("due1", "bob")
|
|
req.DelayMs = &delay
|
|
ctx := context.Background()
|
|
res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if res.State != StateScheduled {
|
|
t.Fatalf("%s", res.State)
|
|
}
|
|
e.setNow(e.nowMs + 60_000)
|
|
n, err := e.app.DispatchDue(ctx, e.nowMs, 10)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n != 1 {
|
|
t.Fatalf("dispatched=%d", n)
|
|
}
|
|
st, _ := e.msgState("alice", "due1")
|
|
if st != StateDispatched {
|
|
t.Fatalf("%s", st)
|
|
}
|
|
})
|
|
|
|
t.Run("F10_recover_clears_pushed_and_extends_grace", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
req := baseSend("rec1", "bob")
|
|
req.Offline = keepTrue()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "rec1")
|
|
// 模拟保留期在停机期间已过
|
|
_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`UPDATE deliveries SET expire_at=?, pushed_conn=? WHERE seq=?`, e.nowMs-1000, "old-conn", seq)
|
|
return err
|
|
})
|
|
_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`UPDATE endpoints SET online_since=?, offline_since=NULL WHERE id=?`, e.nowMs-5000, "bob")
|
|
return err
|
|
})
|
|
start := e.nowMs + 1_000
|
|
e.setNow(start)
|
|
if err := e.app.RecoverOnStart(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var pushed sql.NullString
|
|
var exp sql.NullInt64
|
|
_ = e.db.Read.QueryRow(`SELECT pushed_conn, expire_at FROM deliveries WHERE seq=?`, seq).Scan(&pushed, &exp)
|
|
if pushed.Valid {
|
|
t.Fatalf("pushed_conn still set: %v", pushed.String)
|
|
}
|
|
wantMin := start + 60_000
|
|
if !exp.Valid || exp.Int64 < wantMin {
|
|
t.Fatalf("expire_at=%v want >= %d", exp, wantMin)
|
|
}
|
|
var offline sql.NullInt64
|
|
_ = e.db.Read.QueryRow(`SELECT offline_since FROM endpoints WHERE id=?`, "bob").Scan(&offline)
|
|
if !offline.Valid || offline.Int64 != start {
|
|
t.Fatalf("offline_since=%v want %d", offline, start)
|
|
}
|
|
})
|
|
|
|
t.Run("disabled_target_rejected", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
delay := int64(5_000)
|
|
req := baseSend("dis1", "bob")
|
|
req.DelayMs = &delay
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`UPDATE endpoints SET enabled=0 WHERE id=?`, "bob")
|
|
return err
|
|
})
|
|
e.setNow(e.nowMs + 5_000)
|
|
if _, err := e.app.DispatchDue(ctx, e.nowMs, 10); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "dis1")
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryRejected || reason != ReasonEndpointDisabled {
|
|
t.Fatalf("%s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("queue_full_rejected", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, func(l *Limits) { l.MaxPendingPerReceiver = 1 })
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
req1 := baseSend("qf1", "bob")
|
|
req1.Offline = keepTrue()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
req2 := baseSend("qf2", "bob")
|
|
req2.Offline = keepTrue()
|
|
res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req2)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if res.State != StateCompleted {
|
|
t.Fatalf("state=%s", res.State)
|
|
}
|
|
seq := e.seqOf("alice", "qf2")
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryRejected || reason != ReasonQueueFull {
|
|
t.Fatalf("%s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("sender_left_no_recipients", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
ctx := context.Background()
|
|
_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
|
_, err := tx.Exec(`INSERT INTO groups(id,name,owner_id,created_at) VALUES('g2','g','alice',?)`, e.nowMs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.Exec(`INSERT INTO group_members(group_id,endpoint_id,joined_at) VALUES('g2','alice',?)`, e.nowMs)
|
|
return err
|
|
})
|
|
delay := int64(1000)
|
|
req := &protocol.Send{
|
|
V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "nr1",
|
|
To: protocol.Target{Kind: protocol.TargetGroup, ID: "g2"},
|
|
Body: protocol.Body{Enc: protocol.EncUTF8, Data: "x"}, DelayMs: &delay,
|
|
}
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
e.setNow(e.nowMs + 1000)
|
|
if _, err := e.app.DispatchDue(ctx, e.nowMs, 10); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, reason := e.msgState("alice", "nr1")
|
|
if st != StateCompleted || reason != ReasonNoRecipients {
|
|
t.Fatalf("%s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("too_large_rejected", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.conns.Set("bob", LiveConn{ConnID: "c-bob", MaxReceiveBytes: 50})
|
|
ctx := context.Background()
|
|
req := baseSend("big1", "bob")
|
|
req.Body.Data = string(make([]byte, 200))
|
|
for i := range req.Body.Data {
|
|
// utf8 valid
|
|
_ = i
|
|
}
|
|
req.Body.Data = "{\"x\":\"" + string(make([]byte, 80)) + "\"}"
|
|
// simpler: long ascii
|
|
b := make([]byte, 80)
|
|
for i := range b {
|
|
b[i] = 'a'
|
|
}
|
|
req.Body.Data = string(b)
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "big1")
|
|
st, reason := e.deliveryState(seq, "bob")
|
|
if st != DeliveryRejected || reason != ReasonTooLarge {
|
|
t.Fatalf("%s/%s", st, reason)
|
|
}
|
|
})
|
|
|
|
t.Run("on_publish_dropped_clears_mark", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("drop1", "bob")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
payload := e.down.Snapshots()[0].Payload
|
|
if err := e.app.OnPublishDropped(ctx, "bob", "c-bob", payload); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := e.seqOf("alice", "drop1")
|
|
var pushed sql.NullString
|
|
_ = e.db.Read.QueryRow(`SELECT pushed_conn FROM deliveries WHERE seq=?`, seq).Scan(&pushed)
|
|
if pushed.Valid {
|
|
t.Fatalf("still pushed %s", pushed.String)
|
|
}
|
|
})
|
|
|
|
t.Run("ack_duplicate_and_terminal", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("ack1", "bob")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = e.app.PushPending(ctx, "bob", "c-bob")
|
|
r1, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "1", From: "alice", ID: "ack1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if r1.Result != DeliveryAccepted {
|
|
t.Fatalf("%+v", r1)
|
|
}
|
|
r2, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "2", From: "alice", ID: "ack1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if r2.Result != DeliveryAccepted {
|
|
t.Fatalf("dup %+v", r2)
|
|
}
|
|
})
|
|
|
|
t.Run("disconnect_extends_grace", func(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("dc1", "bob")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = e.app.PushPending(ctx, "bob", "c-bob")
|
|
if err := e.app.OnDisconnect(ctx, "bob", "c-bob", true); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
e.conns.Clear("bob", "c-bob")
|
|
seq := e.seqOf("alice", "dc1")
|
|
var pushed sql.NullString
|
|
var exp sql.NullInt64
|
|
_ = e.db.Read.QueryRow(`SELECT pushed_conn, expire_at FROM deliveries WHERE seq=?`, seq).Scan(&pushed, &exp)
|
|
if pushed.Valid {
|
|
t.Fatal("pushed should clear")
|
|
}
|
|
if !exp.Valid || exp.Int64 != e.nowMs+60_000 {
|
|
t.Fatalf("expire_at=%v", exp)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestRecallDoesNotWriteReceipt(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
ctx := context.Background()
|
|
req := baseSend("nr", "bob")
|
|
req.Offline = keepTrue()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "nr"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var n int
|
|
_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE msg_id=?`, "nr").Scan(&n)
|
|
if n != 0 {
|
|
t.Fatalf("receipts=%d", n)
|
|
}
|
|
}
|
|
|
|
func TestPushRevokedOnRecallAfterPush(t *testing.T) {
|
|
t.Parallel()
|
|
e := openDeliveryEnv(t, nil)
|
|
insertEndpoint(t, e.db, "alice", "", 1, 0)
|
|
insertEndpoint(t, e.db, "bob", "", 1, 0)
|
|
e.online("bob", "c-bob")
|
|
ctx := context.Background()
|
|
if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("rv1", "bob")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = e.app.PushPending(ctx, "bob", "c-bob")
|
|
if _, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "rv1"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
found := false
|
|
for _, p := range e.down.Snapshots() {
|
|
var head struct {
|
|
Type string `json:"type"`
|
|
}
|
|
_ = json.Unmarshal(p.Payload, &head)
|
|
if head.Type == protocol.TypeRevoked {
|
|
found = true
|
|
}
|
|
}
|
|
if !found {
|
|
t.Fatal("expected revoked frame")
|
|
}
|
|
}
|