mirror of
https://github.com/J3vb/OwnCord.git
synced 2026-09-03 03:50:00 +03:00
test(b3-6): seeded hub simulation with a six-part FIFO/seq oracle, fault-injected transport, exact seed replay (Tier 3b/3c) (#1458)
* test(b3-6): seeded hub simulation and fault-injected transport (items 2 and 3) Server/ws/hub_sim_test.go drives a PCG-seeded interleaving of subscribe, broadcast (global, channel, recipients-scoped, sequenced DM), ack, disconnect and reconnect-transfer over a real Hub with eight headless clients, and a model client checks the per-client FIFO/seq oracle from Server/CLAUDE.md after every step: strictly increasing seq per connection, exact audience delivery (nothing lost, extra or twice), a resume replayed exactly from the watermark to the seq at which registerNow ran, h.seq advancing only for a frame that reached the ring, an evicted watermark refused a replay, and a replaced socket's late teardown reporting replaced=true. The resume step runs reconnectRegister as-is (snapshot and registerNow under one seqMu section) on a goroutine while up to three broadcasts race it; the model recovers the snapshot point from the replay burst, so any interleaving is checkable. OWNCORD_SIM_SEED replays one seed, OWNCORD_SIM_SEEDS (default 20) and OWNCORD_SIM_STEPS (default 200) size a run, and a failure prints the seed, the step, a ready-to-paste replay line and the last steps. The default runs in about 2.3 s under -race; `make sim` runs 10,000 steps per seed. Server/ws/faultconn_test.go is the seeded, deterministic frame transport the simulation reads through: drop, tail cut, duplicate, bounded reorder and an order-preserving lag from its own PCG stream, exported to ws_test through export_test.go as NewFaultConnForTest. The simulation's default wire is a lag plus tail cuts, the one fault a TCP-backed WebSocket really has; the silent drop is the negative control that proves the oracle notices a lost replay. BenchmarkReconnectStorm resumes 50 live clients per op through the same path. newTestHub and its three seed helpers take testing.TB so the benchmark can share them. No production code changes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmiqjgTuov1stBTB6uGkvo * docs(b3-6): evidence block for items 2 and 3 (hub simulation, fault transport) Oracle, the RED/GREEN excerpts (inverted assertion, seed replay, drop-all wire, unsynchronized registerNow), wall-clock and benchmark figures, gate results and the epoch-harness decision, under B3-6 in the plan. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmiqjgTuov1stBTB6uGkvo * fix(b3-6): hub sim — deterministic topic limiter for exact replay, a floor on the step mix, auth-frame-wins under transfer, wire-seed mixing Review fixes for items 2 and 3. Exact replay. TopicRateLimiter keys its window on time.Now(), so at 10,000 steps the shed boundary was a timing-dependent step and every later seq differed between runs; the printed OWNCORD_SIM_SEED line could not reproduce a failure. FreezeTopicLimiterForTest (export_test.go) swaps the hub's limiter for one whose window never rolls over inside a run, so the shed is a per-channel count. Three more leaks of the scheduler's interleaving into the trajectory surfaced once that was fixed, and are closed the same way — by taking the decision away from the race or making both outcomes read the same: racing frames are pulled into the wire at attach time (queue fill no longer depends on which side of the snapshot they fell), the racing burst is aimed at the resuming client's own audience (a replay-superset frame was read iff it landed before the snapshot), and a resume within the burst's reach of the ring's eviction boundary is not raced (the allocations could evict the watermark before or after the snapshot and pick replay or fallback). Three runs of one seed now print byte-identical stats; what still varies — how many racing seqs land in the replay burst — is printed on its own line and stated in the doc comment. Floor. TestHubSimulation aggregates the per-seed stats and requires every load-bearing transition (the four broadcast kinds, resume, fallback, fresh, cut, kicked, racing-in-replay) at least once across the default run, so a constant change cannot turn the simulation into no-ops with CI green. Its first run found that the overflow kick had become unreachable at 200 steps; the sim's queue is 12 now (production stays 256). Also: the resume step draws active_channel_id as none / the open channel / another channel whether or not the old socket is registered, so registerNow's auth-frame-wins branch runs under the transfer; the wire's PCG takes the seed and (idx<<32|conns) as its two words instead of an arithmetic mix that collided past 131 connections; seedTestUser takes testing.TB like its siblings; the evidence block lists what the simulation does not cover and the -timeout 60m the ten-pass deadlock gate needs, with the same line under Traps carried forward. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmiqjgTuov1stBTB6uGkvo * test(b3-6): hub sim — floor on raced resumes instead of the scheduler-decided key; log when the floor is skipped racing-in-replay was the one floor key the scheduler decides, so a correct hub could in principle fail the floor on a run where no racing seq landed inside a burst. The floor now keys on raced resumes — a resume that got a replay while a burst ran (burst > 0 && ok), which the seed determines — and racing-in-replay stays a printed count. The floor also says so when it is skipped for OWNCORD_SIM_SEED or a shorter seed list instead of returning silently. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmiqjgTuov1stBTB6uGkvo * test(b3-6): hub sim — raced counts a broadcast allocated while registration was in progress, not the requested burst Codex P2 on #1458: raced incremented on burst > 0, which counted a resume whose goroutine had returned before the first broadcast ran and one whose every channel broadcast the limiter shed, so the floor could pass with no broadcast overlapping a registration. raced now counts a resume where a racing broadcast allocated a seq while the reconnect goroutine had not yet been observed to return (the driver's done handshake, checked after each allocation). That is the scheduler's call, so raced moves off the deterministic stats line and is floored only in aggregate across the 20 default seeds — 179 bursts per run, 178–179 observed overlapping in three measured runs, odds named in the comment and the evidence block. The requested burst stays a printed, seed-determined count (bursts) and is floored as before; the floor logs its totals. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmiqjgTuov1stBTB6uGkvo --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
J3vb
+12
-1
@@ -4,6 +4,8 @@
|
||||
# test-deadlock Run the deadlock-detection pass CI also runs.
|
||||
# fuzz Actually fuzz. CI (and plain `go test`) only replays the
|
||||
# committed seed corpus; this generates new inputs.
|
||||
# sim Run the seeded hub simulation long: 10,000 steps per seed.
|
||||
# CI runs its 200 x 20 default through `go test -race ./...`.
|
||||
# cover Per-package coverage (what CI uploads) + a function summary.
|
||||
# cover-all Cross-package coverage — the honest number. See below.
|
||||
# sqlc-generate Regenerate type-safe Go from sqlc.yaml (db/dbgen).
|
||||
@@ -18,7 +20,7 @@
|
||||
|
||||
SQLC_VERSION := $(shell cat sqlc.version)
|
||||
|
||||
.PHONY: test test-deadlock fuzz cover cover-all sqlc-install sqlc-generate sqlc-verify \
|
||||
.PHONY: test test-deadlock fuzz sim cover cover-all sqlc-install sqlc-generate sqlc-verify \
|
||||
protocol-generate protocol-verify docs-generate docs-verify otel-up otel-down
|
||||
|
||||
test:
|
||||
@@ -50,6 +52,15 @@ fuzz:
|
||||
done; \
|
||||
done
|
||||
|
||||
# The seeded hub simulation (ws/hub_sim_test.go), long form: the default 20
|
||||
# seeds at 10,000 steps each instead of the 200 CI runs. A failure prints a
|
||||
# ready-to-paste OWNCORD_SIM_SEED=... OWNCORD_SIM_STEPS=... replay line.
|
||||
#
|
||||
# No make on Windows? OWNCORD_SIM_STEPS=10000 go test -race -count=1 -run '^TestHubSimulation$' ./ws/
|
||||
SIMSTEPS ?= 10000
|
||||
sim:
|
||||
OWNCORD_SIM_STEPS=$(SIMSTEPS) go test -race -count=1 -run '^TestHubSimulation$$' ./ws/
|
||||
|
||||
# Matches the CI invocation. Note that `go test ./... -coverprofile` instruments
|
||||
# each package only for itself, so a package whose code is mostly exercised
|
||||
# through another package's tests reports far lower than its real coverage
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os/exec"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/J3vb/OwnCord/Server/db"
|
||||
@@ -473,3 +474,97 @@ func (h *Hub) BroadcastVoiceEventForTest(channelID int64, msg []byte) {
|
||||
// MaxColdReplayForTest exposes the cold-tier replay row cap so tests can seed
|
||||
// exactly enough events to hit it.
|
||||
const MaxColdReplayForTest = maxColdReplay
|
||||
|
||||
// ─── hub simulation helpers (hub_sim_test.go, B3-6) ────────────────────────
|
||||
|
||||
// NewSimClientForTest builds a headless client the way newClient does for a
|
||||
// real socket — separate normal/high/low queues and a resume watermark — minus
|
||||
// the conn. channelID is the auth-frame active_channel_id handleReconnect
|
||||
// promotes after checking it against the allowed set (0 = none).
|
||||
func NewSimClientForTest(hub *Hub, user *db.User, channelID int64, lastSeq uint64, send, sendHigh, sendLow chan []byte) *Client {
|
||||
return &Client{
|
||||
hub: hub,
|
||||
ctx: context.Background(),
|
||||
userID: user.ID,
|
||||
user: user,
|
||||
channelID: channelID,
|
||||
lastSeq: lastSeq,
|
||||
send: send,
|
||||
sendHigh: sendHigh,
|
||||
sendLow: sendLow,
|
||||
}
|
||||
}
|
||||
|
||||
// DeliverBroadcastForTest runs deliverBroadcast synchronously on the caller's
|
||||
// goroutine — the dispatch loop's critical section without the dispatch loop —
|
||||
// and returns the seq it allocated, or 0 when the topic limiter shed the
|
||||
// frame. A non-nil recipients selects the visibility-filtered branch
|
||||
// (voice_state's path). The seq is read off h.seq before and after, so it is
|
||||
// exact only while the caller is the sole allocator, which the simulation
|
||||
// guarantees.
|
||||
func (h *Hub) DeliverBroadcastForTest(channelID int64, recipients []int64, msg []byte) uint64 {
|
||||
before := atomic.LoadUint64(&h.seq)
|
||||
h.deliverBroadcast(broadcastMsg{channelID: channelID, recipients: recipients, msg: msg})
|
||||
if after := atomic.LoadUint64(&h.seq); after != before {
|
||||
return after
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// SendSequencedToUsersForTest exposes sendSequencedToUsers (the sequenced DM
|
||||
// path) and returns the seq it allocated, under the same sole-allocator caveat
|
||||
// as DeliverBroadcastForTest.
|
||||
func (h *Hub) SendSequencedToUsersForTest(channelID int64, userIDs []int64, msg []byte) uint64 {
|
||||
before := atomic.LoadUint64(&h.seq)
|
||||
h.sendSequencedToUsers(channelID, userIDs, msg)
|
||||
if after := atomic.LoadUint64(&h.seq); after != before {
|
||||
return after
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// ReconnectRegisterForTest exposes reconnectRegister's buffer-tier path: the
|
||||
// replay snapshot and registerNow inside ONE h.seqMu critical section, exactly
|
||||
// as handleReconnect runs it. ok=false means the ring no longer covers lastSeq
|
||||
// and production would fall through to a full ready.
|
||||
func (h *Hub) ReconnectRegisterForTest(c *Client, lastSeq uint64, allowed map[int64]bool) ([][]byte, bool) {
|
||||
return h.reconnectRegister(context.Background(), c, lastSeq, allowed, "buffer", nil, 0)
|
||||
}
|
||||
|
||||
// UnregisterNowForTest exposes unregisterNow; the return is its "replaced"
|
||||
// verdict (true when a newer connection holds the slot).
|
||||
func (h *Hub) UnregisterNowForTest(c *Client) bool {
|
||||
return h.unregisterNow(c)
|
||||
}
|
||||
|
||||
// SeqForTest reads the hub's monotonic seq counter.
|
||||
func (h *Hub) SeqForTest() uint64 {
|
||||
return atomic.LoadUint64(&h.seq)
|
||||
}
|
||||
|
||||
// IsSendClosedForTest exposes Client.isSendClosed.
|
||||
func IsSendClosedForTest(c *Client) bool {
|
||||
return c.isSendClosed()
|
||||
}
|
||||
|
||||
// CloseSendForTest exposes Client.closeSend, the pump teardown's last step.
|
||||
func CloseSendForTest(c *Client) {
|
||||
c.closeSend()
|
||||
}
|
||||
|
||||
// NewFaultConnForTest builds the fault-injecting frame transport of
|
||||
// faultconn_test.go over preface (delivered first, e.g. a replay burst) and
|
||||
// then in (a client's outbound queue; nil for a preface-only source). seed
|
||||
// and stream are the PCG's two words.
|
||||
func NewFaultConnForTest(seed, stream uint64, sched FaultSchedule, preface [][]byte, in <-chan []byte) *FaultConn {
|
||||
return newFaultConn(seed, stream, sched, preface, in)
|
||||
}
|
||||
|
||||
// FreezeTopicLimiterForTest swaps the hub's per-topic limiter for one whose
|
||||
// window never rolls over inside a test run, so a shed is a deterministic
|
||||
// count — the first topicRateLimitPerSecond frames per channel pass, the rest
|
||||
// shed — instead of one that depends on where time.Now() falls. Call before
|
||||
// the first broadcast; the simulation needs it so a seed replays exactly.
|
||||
func (h *Hub) FreezeTopicLimiterForTest() {
|
||||
h.topicLimiter = NewTopicRateLimiter(topicRateLimitPerSecond, time.Hour)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,314 @@
|
||||
package ws
|
||||
|
||||
// faultconn_test.go — B3-6 item 3: a seeded, deterministic fault-injecting
|
||||
// frame transport (Tier 3c of docs/plans/bug-detection-improvements.md).
|
||||
//
|
||||
// FaultConn sits at the layer the headless-client harness actually uses: it
|
||||
// wraps a client's outbound queue (the chan a real socket's writePump would
|
||||
// drain), optionally preceded by a preface of handshake frames such as a
|
||||
// resume's replay burst, and re-emits the frames with drops, duplicates,
|
||||
// bounded reordering and an order-preserving lag drawn from its own PCG
|
||||
// stream. Same seed, same schedule, same input order: same output.
|
||||
//
|
||||
// The one fault a TCP-backed WebSocket can really produce is a cut — every
|
||||
// frame after some point is lost and the client must resume from its
|
||||
// watermark. That is DropTail. Silent per-frame drops, duplicates and
|
||||
// reorders are impossible on the real wire; they exist so a harness can
|
||||
// prove its oracle notices them (hub_sim_test.go's RED control) and for the
|
||||
// client model test, whose store must tolerate whatever a reconnect replays.
|
||||
//
|
||||
// ws_test reaches it through export_test.go (NewFaultConnForTest); the types
|
||||
// below are exported so the schedule can be written from there too.
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"math/rand/v2"
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// FaultSchedule is the per-frame fault mix a FaultConn applies.
|
||||
type FaultSchedule struct {
|
||||
Drop float64 // per-frame probability the frame is lost
|
||||
DropTail bool // a drop is a socket death: frames still in flight are lost and Recv reports FaultCut
|
||||
Dup float64 // per-frame probability the frame is delivered twice, back to back
|
||||
Reorder float64 // per-frame probability the frame is pushed back up to ReorderWindow places
|
||||
ReorderWindow int // bound for Reorder; 0 means 1
|
||||
Delay int // lag in frames: a frame is released only once Delay later frames were pulled, or the source closed
|
||||
}
|
||||
|
||||
// FaultStatus is what Recv reports alongside a frame.
|
||||
type FaultStatus int
|
||||
|
||||
const (
|
||||
FaultOK FaultStatus = iota // a frame
|
||||
FaultEmpty // nothing releasable yet; the source is still open
|
||||
FaultClosed // the source closed and everything it wrote has been released
|
||||
FaultCut // the schedule cut the connection; whatever was still in flight is gone
|
||||
)
|
||||
|
||||
type faultFrame struct {
|
||||
key, idx int
|
||||
data []byte
|
||||
}
|
||||
|
||||
// FaultConn is the transport. Not safe for concurrent use; one reader owns it.
|
||||
type FaultConn struct {
|
||||
rng *rand.Rand
|
||||
sched FaultSchedule
|
||||
preface [][]byte
|
||||
in <-chan []byte
|
||||
pending []faultFrame // pulled, faults applied, awaiting release; sorted by key
|
||||
pulled int
|
||||
closed bool
|
||||
cut bool
|
||||
|
||||
Dropped, Duplicated, Reordered int
|
||||
}
|
||||
|
||||
// seed and stream are the PCG's two words — one seed per run, one stream per
|
||||
// connection — so no arithmetic mix can make two connections collide.
|
||||
func newFaultConn(seed, stream uint64, sched FaultSchedule, preface [][]byte, in <-chan []byte) *FaultConn {
|
||||
return &FaultConn{
|
||||
rng: rand.New(rand.NewPCG(seed, stream)),
|
||||
sched: sched,
|
||||
preface: preface,
|
||||
in: in,
|
||||
}
|
||||
}
|
||||
|
||||
// pull moves every frame the source has ready into pending, drawing the
|
||||
// schedule per frame in pull order — so batching never changes the outcome.
|
||||
func (f *FaultConn) pull() {
|
||||
for !f.cut && !f.closed {
|
||||
var data []byte
|
||||
switch {
|
||||
case len(f.preface) > 0:
|
||||
data, f.preface = f.preface[0], f.preface[1:]
|
||||
case f.in == nil:
|
||||
f.closed = true
|
||||
return
|
||||
default:
|
||||
select {
|
||||
case d, ok := <-f.in:
|
||||
if !ok {
|
||||
f.closed = true
|
||||
return
|
||||
}
|
||||
data = d
|
||||
default:
|
||||
return
|
||||
}
|
||||
}
|
||||
idx := f.pulled
|
||||
f.pulled++
|
||||
if f.rng.Float64() < f.sched.Drop {
|
||||
f.Dropped++
|
||||
f.cut = f.sched.DropTail
|
||||
continue
|
||||
}
|
||||
key := idx
|
||||
if f.rng.Float64() < f.sched.Reorder {
|
||||
key += 1 + f.rng.IntN(max(f.sched.ReorderWindow, 1))
|
||||
f.Reordered++
|
||||
}
|
||||
f.pending = append(f.pending, faultFrame{key: key, idx: idx, data: data})
|
||||
if f.rng.Float64() < f.sched.Dup {
|
||||
f.Duplicated++
|
||||
f.pending = append(f.pending, faultFrame{key: key, idx: idx, data: data})
|
||||
}
|
||||
slices.SortStableFunc(f.pending, func(a, b faultFrame) int { return cmp.Compare(a.key, b.key) })
|
||||
}
|
||||
}
|
||||
|
||||
// Recv returns the next frame the transport delivers. A closed source flushes
|
||||
// the lag (the server wrote those frames before closing; TCP delivers them); a
|
||||
// cut does not (they were in flight when the socket died).
|
||||
func (f *FaultConn) Recv() ([]byte, FaultStatus) {
|
||||
f.pull()
|
||||
if len(f.pending) > 0 {
|
||||
head := f.pending[0]
|
||||
if f.closed || f.pulled > head.key+f.sched.Delay {
|
||||
f.pending = f.pending[1:]
|
||||
return head.data, FaultOK
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case f.cut:
|
||||
f.pending = nil
|
||||
return nil, FaultCut
|
||||
case f.closed:
|
||||
return nil, FaultClosed
|
||||
}
|
||||
return nil, FaultEmpty
|
||||
}
|
||||
|
||||
// Buffered counts the frames the transport still holds — preface not yet
|
||||
// pulled plus pulled frames awaiting release — none of which the reader has
|
||||
// seen and none of which are lost.
|
||||
func (f *FaultConn) Buffered() int {
|
||||
return len(f.preface) + len(f.pending)
|
||||
}
|
||||
|
||||
// Cut reports whether the schedule has already cut the connection: the reader
|
||||
// may still drain the frames whose lag the cut satisfied, then Recv reports
|
||||
// FaultCut. Anything the source writes from now on is lost.
|
||||
func (f *FaultConn) Cut() bool {
|
||||
return f.cut
|
||||
}
|
||||
|
||||
// Pull moves whatever the source holds right now into the transport, faults
|
||||
// applied, releasing nothing. A harness whose queue-fill accounting must not
|
||||
// depend on when frames were written (hub_sim_test.go's attach) calls it.
|
||||
func (f *FaultConn) Pull() {
|
||||
f.pull()
|
||||
}
|
||||
|
||||
// ─── tests ───────────────────────────────────────────────────────────────────
|
||||
|
||||
func faultFrames(n int) [][]byte {
|
||||
out := make([][]byte, n)
|
||||
for i := range n {
|
||||
out[i] = []byte{byte(i)}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func faultDrain(f *FaultConn) (frames []byte, last FaultStatus) {
|
||||
for {
|
||||
data, st := f.Recv()
|
||||
if st != FaultOK {
|
||||
return frames, st
|
||||
}
|
||||
frames = append(frames, data[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestFaultConn_IdentityAndDeterminism(t *testing.T) {
|
||||
in := faultFrames(20)
|
||||
got, st := faultDrain(newFaultConn(1, 0, FaultSchedule{}, in, nil))
|
||||
if st != FaultClosed || len(got) != 20 {
|
||||
t.Fatalf("identity schedule: %d frames, status %d; want 20, FaultClosed", len(got), st)
|
||||
}
|
||||
for i, b := range got {
|
||||
if int(b) != i {
|
||||
t.Fatalf("identity schedule reordered: frame %d is %d", i, b)
|
||||
}
|
||||
}
|
||||
|
||||
sched := FaultSchedule{Drop: 0.2, Dup: 0.2, Reorder: 0.3, ReorderWindow: 3, Delay: 1}
|
||||
a, _ := faultDrain(newFaultConn(7, 1, sched, in, nil))
|
||||
b, _ := faultDrain(newFaultConn(7, 1, sched, in, nil))
|
||||
c, _ := faultDrain(newFaultConn(8, 1, sched, in, nil))
|
||||
d, _ := faultDrain(newFaultConn(7, 2, sched, in, nil))
|
||||
if !slices.Equal(a, b) {
|
||||
t.Fatalf("same seed and stream, different output:\n%v\n%v", a, b)
|
||||
}
|
||||
if slices.Equal(a, c) || slices.Equal(a, d) {
|
||||
t.Fatalf("a different seed (7 vs 8) or stream (1 vs 2) produced the same schedule %v", a)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFaultConn_Schedule(t *testing.T) {
|
||||
in := faultFrames(20)
|
||||
|
||||
t.Run("drop all", func(t *testing.T) {
|
||||
f := newFaultConn(1, 0, FaultSchedule{Drop: 1}, in, nil)
|
||||
got, st := faultDrain(f)
|
||||
if len(got) != 0 || st != FaultClosed || f.Dropped != 20 {
|
||||
t.Fatalf("got %d frames, status %d, dropped %d; want 0, FaultClosed, 20", len(got), st, f.Dropped)
|
||||
}
|
||||
})
|
||||
t.Run("drop tail cuts", func(t *testing.T) {
|
||||
f := newFaultConn(3, 0, FaultSchedule{Drop: 0.15, DropTail: true}, in, nil)
|
||||
got, st := faultDrain(f)
|
||||
if st != FaultCut || f.Dropped != 1 || len(got) >= 20 {
|
||||
t.Fatalf("got %d frames, status %d, dropped %d; want a prefix, FaultCut, 1", len(got), st, f.Dropped)
|
||||
}
|
||||
for i, b := range got {
|
||||
if int(b) != i {
|
||||
t.Fatalf("prefix before the cut is not in order: %v", got)
|
||||
}
|
||||
}
|
||||
if _, again := f.Recv(); again != FaultCut {
|
||||
t.Fatalf("a cut connection must stay cut, got %d", again)
|
||||
}
|
||||
})
|
||||
t.Run("dup", func(t *testing.T) {
|
||||
f := newFaultConn(1, 0, FaultSchedule{Dup: 1}, in, nil)
|
||||
got, _ := faultDrain(f)
|
||||
if len(got) != 40 || f.Duplicated != 20 {
|
||||
t.Fatalf("got %d frames, duplicated %d; want 40, 20", len(got), f.Duplicated)
|
||||
}
|
||||
for i := 0; i < 40; i += 2 {
|
||||
if got[i] != got[i+1] || int(got[i]) != i/2 {
|
||||
t.Fatalf("duplicates are not adjacent and in order: %v", got)
|
||||
}
|
||||
}
|
||||
})
|
||||
t.Run("reorder is a bounded permutation", func(t *testing.T) {
|
||||
// Half the frames: pushing every frame back keeps their keys sorted
|
||||
// and the order intact — a reorder needs an unpushed neighbour.
|
||||
f := newFaultConn(5, 0, FaultSchedule{Reorder: 0.5, ReorderWindow: 2}, in, nil)
|
||||
got, _ := faultDrain(f)
|
||||
sorted := slices.Clone(got)
|
||||
slices.Sort(sorted)
|
||||
if !slices.Equal(sorted, in2bytes(in)) {
|
||||
t.Fatalf("reorder lost or invented frames: %v", got)
|
||||
}
|
||||
if slices.Equal(got, in2bytes(in)) || f.Reordered == 0 {
|
||||
t.Fatalf("Reorder=0.5 left the order intact (%d pushed): %v", f.Reordered, got)
|
||||
}
|
||||
for i, b := range got {
|
||||
if int(b) > i+2 || int(b) < i-2 {
|
||||
t.Fatalf("frame %d moved more than the window: position %d", b, i)
|
||||
}
|
||||
}
|
||||
})
|
||||
t.Run("delay lags an open source and flushes on close", func(t *testing.T) {
|
||||
ch := make(chan []byte, 8)
|
||||
f := newFaultConn(1, 0, FaultSchedule{Delay: 2}, nil, ch)
|
||||
for _, fr := range faultFrames(5) {
|
||||
ch <- fr
|
||||
}
|
||||
got, st := faultDrain(f)
|
||||
if st != FaultEmpty || len(got) != 3 || f.Buffered() != 2 {
|
||||
t.Fatalf("open source: got %d frames, status %d, buffered %d; want 3, FaultEmpty, 2", len(got), st, f.Buffered())
|
||||
}
|
||||
close(ch)
|
||||
rest, st := faultDrain(f)
|
||||
if st != FaultClosed || len(rest) != 2 || rest[0] != 3 || rest[1] != 4 {
|
||||
t.Fatalf("closed source: got %v, status %d; want [3 4], FaultClosed", rest, st)
|
||||
}
|
||||
})
|
||||
t.Run("cut loses the lag", func(t *testing.T) {
|
||||
// One frame per Recv so the schedule can be switched between pulls:
|
||||
// frames 0 and 1 arrive intact, frame 2 is the cut. With Delay 2 the
|
||||
// cut satisfies frame 0's lag but not frame 1's — it was in flight.
|
||||
ch := make(chan []byte, 1)
|
||||
f := newFaultConn(1, 0, FaultSchedule{Delay: 2, DropTail: true}, nil, ch)
|
||||
var got []byte
|
||||
for i := range 3 {
|
||||
f.sched.Drop = 0
|
||||
if i == 2 {
|
||||
f.sched.Drop = 1
|
||||
}
|
||||
ch <- []byte{byte(i)}
|
||||
if data, st := f.Recv(); st == FaultOK {
|
||||
got = append(got, data[0])
|
||||
}
|
||||
}
|
||||
_, st := f.Recv()
|
||||
if st != FaultCut || !slices.Equal(got, []byte{0}) {
|
||||
t.Fatalf("got %v, status %d; want [0] then FaultCut (frame 1 was in flight)", got, st)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func in2bytes(in [][]byte) []byte {
|
||||
out := make([]byte, len(in))
|
||||
for i, fr := range in {
|
||||
out[i] = fr[0]
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package ws_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/J3vb/OwnCord/Server/db"
|
||||
"github.com/J3vb/OwnCord/Server/ws"
|
||||
)
|
||||
|
||||
// BenchmarkReconnectStorm is hub_sim_test.go's reconnect-transfer step over 50
|
||||
// live clients: one op resumes every client once, each resume preceded by a
|
||||
// global broadcast so its 8-frame replay is real and the registration races
|
||||
// nothing but the seqMu it takes. Item 6 collects the baseline; run with
|
||||
//
|
||||
// go test -run '^$' -bench ReconnectStorm -benchmem ./ws/
|
||||
func BenchmarkReconnectStorm(b *testing.B) {
|
||||
hub, database := newTestHub(b)
|
||||
const n = 50
|
||||
payload := []byte(`{"type":"sim","bench":true}`)
|
||||
newConn := func(u *db.User, lastSeq uint64) *ws.Client {
|
||||
return ws.NewSimClientForTest(hub, u, 0, lastSeq, make(chan []byte, 256), make(chan []byte, 64), make(chan []byte, 64))
|
||||
}
|
||||
users := make([]*db.User, n)
|
||||
conns := make([]*ws.Client, n)
|
||||
for i := range n {
|
||||
users[i] = seedOwnerUser(b, database, fmt.Sprintf("storm-%d", i))
|
||||
conns[i] = newConn(users[i], 0)
|
||||
hub.RegisterNowForTest(conns[i])
|
||||
}
|
||||
for range 16 { // so every watermark below sits inside the ring
|
||||
hub.DeliverBroadcastForTest(0, nil, payload)
|
||||
}
|
||||
allowed := map[int64]bool{}
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for range b.N {
|
||||
for i := range n {
|
||||
seq := hub.DeliverBroadcastForTest(0, nil, payload)
|
||||
nc := newConn(users[i], seq-8)
|
||||
if _, ok := hub.ReconnectRegisterForTest(nc, seq-8, allowed); !ok {
|
||||
b.Fatalf("resume from %d refused a replay", seq-8)
|
||||
}
|
||||
hub.UnregisterNowForTest(conns[i])
|
||||
conns[i] = nc
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
package ws_test
|
||||
|
||||
// hub_sim_oracle_test.go — the checks behind hub_sim_test.go's oracle; the
|
||||
// invariants I1–I6 are stated there. Split out so the driver stays readable;
|
||||
// nothing here runs on its own.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"slices"
|
||||
|
||||
"github.com/J3vb/OwnCord/Server/ws"
|
||||
)
|
||||
|
||||
// observe is I1 and I2 for one frame read off a connection's normal queue.
|
||||
func (s *sim) observe(c *simClient, frame []byte) {
|
||||
seq := simSeqOf(frame)
|
||||
switch {
|
||||
case seq == 0:
|
||||
s.failf("I1: unsequenced frame on c%d's normal queue: %s", c.idx, frame)
|
||||
case seq <= c.w:
|
||||
s.failf("I1: c%d conn %d: seq %d after watermark %d", c.idx, c.conns, seq, c.w)
|
||||
case len(c.owed) == 0:
|
||||
s.failf("I2: c%d conn %d: received seq %d but is owed nothing", c.idx, c.conns, seq)
|
||||
case c.owed[0] != seq:
|
||||
s.failf("I2: c%d conn %d: expected seq %d next, got %d (owed %v)", c.idx, c.conns, c.owed[0], seq, c.owed)
|
||||
}
|
||||
c.owed = c.owed[1:]
|
||||
c.w = seq
|
||||
}
|
||||
|
||||
func simSeqOf(frame []byte) uint64 {
|
||||
var env struct {
|
||||
Seq uint64 `json:"seq"`
|
||||
}
|
||||
if json.Unmarshal(frame, &env) != nil {
|
||||
return 0
|
||||
}
|
||||
return env.Seq
|
||||
}
|
||||
|
||||
// checkPriorityQueues is I1's second half: nothing sequenced may sit on the
|
||||
// high or low queue, where writePump would let it overtake the FIFO.
|
||||
func (s *sim) checkPriorityQueues(c *simClient) {
|
||||
for _, q := range []chan []byte{c.high, c.low} {
|
||||
for drained := false; !drained; {
|
||||
select {
|
||||
case frame, ok := <-q:
|
||||
if ok && simSeqOf(frame) != 0 {
|
||||
s.failf("I1: seq-stamped frame on c%d's priority queue: %s", c.idx, frame)
|
||||
}
|
||||
drained = !ok
|
||||
default:
|
||||
drained = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkReplay is I3. S is unknown when broadcasts raced the registration, so
|
||||
// every candidate — the seq before the burst, then each racing seq — is tried;
|
||||
// the replay burst is a prefix-closed function of S, so at most one matches,
|
||||
// and the racing seqs above it are what the new connection is owed live.
|
||||
func (s *sim) checkReplay(c *simClient, events [][]byte, s0 uint64, racing []simAlloc) []uint64 {
|
||||
got := make([]uint64, len(events))
|
||||
for i, e := range events {
|
||||
got[i] = simSeqOf(e)
|
||||
}
|
||||
candidates := make([]uint64, 0, 1+len(racing))
|
||||
candidates = append(candidates, s0)
|
||||
for _, a := range racing {
|
||||
candidates = append(candidates, a.seq)
|
||||
}
|
||||
for j, snap := range candidates {
|
||||
if !slices.Equal(got, s.replayExpected(c, snap)) {
|
||||
continue
|
||||
}
|
||||
owed := slices.Clone(got)
|
||||
for _, a := range racing[j:] {
|
||||
if a.reaches(c) {
|
||||
owed = append(owed, a.seq)
|
||||
}
|
||||
}
|
||||
if j > 0 {
|
||||
s.racing++
|
||||
}
|
||||
return owed
|
||||
}
|
||||
s.failf("I3: c%d resume from W=%d: replay %v matches no snapshot point; at S=%d it should be %v (racing seqs after that: %v)",
|
||||
c.idx, c.w, got, s0, s.replayExpected(c, s0), candidates[1:])
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *sim) replayExpected(c *simClient, snap uint64) []uint64 {
|
||||
exp := []uint64{}
|
||||
for seq := c.w + 1; seq <= snap; seq++ {
|
||||
if ch := s.chanOf[seq]; ch == 0 || c.allowed[ch] {
|
||||
exp = append(exp, seq)
|
||||
}
|
||||
}
|
||||
return exp
|
||||
}
|
||||
|
||||
// checkCounts is I2's after-every-step half: every live connection holds
|
||||
// exactly the frames it is owed, no more and no fewer. A connection the wire
|
||||
// has already cut (the client just has not read up to the cut yet) and one
|
||||
// the server closed (overflow kick, read as FaultClosed later) are dead: the
|
||||
// next read settles them, and a dead socket owes nothing.
|
||||
func (s *sim) checkCounts() {
|
||||
for _, c := range s.clients {
|
||||
if c.conn == nil || c.cut || c.wire.Cut() || ws.IsSendClosedForTest(c.conn) {
|
||||
continue
|
||||
}
|
||||
if unread := len(c.send) + c.wire.Buffered(); unread != len(c.owed) {
|
||||
s.failf("I2: c%d conn %d holds %d unread frame(s) but is owed %d: %v", c.idx, c.conns, unread, len(c.owed), c.owed)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,624 @@
|
||||
package ws_test
|
||||
|
||||
// hub_sim_test.go — B3-6 item 2: a seeded simulation of the hub's ordering
|
||||
// space (Tier 3b of docs/plans/bug-detection-improvements.md). Every recurring
|
||||
// bug in this package's history is a bad ordering — registerNow's
|
||||
// reconnect-transfer, replay against a moving ring, a socket dying with
|
||||
// frames in flight — and nothing else in the repo generates orderings. This
|
||||
// does: a PCG stream picks each step from {subscribe, broadcast, ack,
|
||||
// disconnect, reconnect-transfer} over eight headless clients on a real Hub,
|
||||
// and a model client checks the oracle below after every step.
|
||||
//
|
||||
// ORACLE — written before the driver, from Server/CLAUDE.md ("Sequenced
|
||||
// frames share one per-client FIFO because clients ack only max(seq) — a
|
||||
// frame that skips the queue, or a seq allocated for a frame that is then
|
||||
// dropped, is silently unrecoverable") and the replay semantics pinned by
|
||||
// ringbuffer_test.go, hub_register_race_test.go and serve.go's
|
||||
// reconnectRegister. The only ack this protocol has is last_seq on the next
|
||||
// auth frame, so "ack" here is the client reading frames and advancing its
|
||||
// watermark W = max(seq read).
|
||||
//
|
||||
// I1 Within one connection, sequenced frames arrive with strictly
|
||||
// increasing seq — the replay burst first, then the live queue — and a
|
||||
// seq-stamped frame never rides the high- or low-priority queue.
|
||||
// I2 A live connection yields, in order, exactly the seqs the hub allocated
|
||||
// for events whose audience contained it since it registered: nothing
|
||||
// missing (a seq allocated for a frame that was then dropped is a gap),
|
||||
// nothing extra, nothing twice. After every step the frames still
|
||||
// unread (queue + transport) equal the seqs still owed.
|
||||
// I3 On a resume from W that the ring covers, the replay burst is exactly
|
||||
// {s in (W, S] : channel(s) is 0 or READ-allowed}, in seq order, where
|
||||
// S is the hub seq at the instant registerNow ran (atomic with the
|
||||
// snapshot under h.seqMu), and every audience seq above S arrives live —
|
||||
// so across a drop and a resume no audience seq above W is lost or
|
||||
// skipped. Frames the dying socket still held above W are delivered
|
||||
// again by the replay; that cross-connection duplicate is by design
|
||||
// (the client acks max(seq)) and I1 still holds per connection.
|
||||
// I4 h.seq advances only for a frame that reached the ring: a
|
||||
// topic-limiter shed allocates nothing, and after every allocation
|
||||
// ring.NewestSeq() == h.seq.
|
||||
// I5 When the ring cannot cover W the resume is refused a replay
|
||||
// (ok=false), the connection is registered the way handleFreshConnect
|
||||
// registers a replay-failure fallback, and the obligations restart
|
||||
// there — a full ready carries the state.
|
||||
// I6 A dead socket's late teardown (unregisterNow after its replacement
|
||||
// registered) reports replaced=true and leaves the replacement in
|
||||
// place; tearing down the live connection reports false.
|
||||
//
|
||||
// Audience: a global broadcast reaches every registered client; a channel
|
||||
// broadcast the clients focused on it (after the topic limiter); a
|
||||
// recipients-scoped broadcast (voice_state's path) exactly its listed users;
|
||||
// a sequenced DM its participants. The replay filter is coarser than the
|
||||
// audience — it replays every READ-allowed channel, focused or not — so a
|
||||
// resume legitimately carries frames the client would not have received live.
|
||||
//
|
||||
// Knobs: OWNCORD_SIM_SEED replays one seed; OWNCORD_SIM_SEEDS (default 20)
|
||||
// and OWNCORD_SIM_STEPS (default 200) size a run; `make sim` runs 10,000
|
||||
// steps. A failure prints the seed, the step, a ready-to-paste replay line
|
||||
// and the last steps.
|
||||
//
|
||||
// Determinism: the step sequence, every seq allocation (the topic limiter is
|
||||
// frozen to a per-channel count, FreezeTopicLimiterForTest), every wire fault
|
||||
// and every figure on the stats line are a pure function of the seed — three
|
||||
// runs of one seed print byte-identical stats — so a failure in any of them
|
||||
// replays exactly from the printed line. What still varies between runs is
|
||||
// the scheduler's interleaving inside a racing reconnect step: which of the
|
||||
// racing seqs land in the replay burst and which arrive live. The oracle
|
||||
// holds for every interleaving, and the burst is shaped so the client reads
|
||||
// the same frames either way (see broadcast), but a defect that depends on
|
||||
// that split — the I3 class — replays only probabilistically; run the seed
|
||||
// with -count. The default run also carries a floor (TestHubSimulation):
|
||||
// every load-bearing transition must occur at least once across the seeds,
|
||||
// so a constant change cannot quietly turn the simulation into no-ops — and
|
||||
// one of them, a broadcast observed running against a registration in
|
||||
// flight, is the scheduler's and is floored only in aggregate.
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand/v2"
|
||||
"os"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/J3vb/OwnCord/Server/db"
|
||||
"github.com/J3vb/OwnCord/Server/ws"
|
||||
)
|
||||
|
||||
const (
|
||||
simClients = 8
|
||||
simChannels = 3
|
||||
simRing = 48 // replay ring (production 1000): small enough that I5's eviction is reachable in 200 steps
|
||||
simSendBuf = 12 // normal queue (production 256, client.go sendBufSize): small enough that the BUG-124 overflow kick happens a few times per seed at 200 steps
|
||||
simDefaultSeeds = 20
|
||||
simDefaultSteps = 200
|
||||
)
|
||||
|
||||
type simOp int
|
||||
|
||||
const (
|
||||
opSubscribe simOp = iota
|
||||
opBroadcast
|
||||
opAck
|
||||
opDisconnect
|
||||
opReconnect
|
||||
)
|
||||
|
||||
var simOpWeights = [...]int{opSubscribe: 15, opBroadcast: 40, opAck: 20, opDisconnect: 10, opReconnect: 15}
|
||||
|
||||
type simKind int
|
||||
|
||||
const (
|
||||
kindGlobal simKind = iota
|
||||
kindChannel
|
||||
kindRecipients
|
||||
kindDM
|
||||
)
|
||||
|
||||
var (
|
||||
simKindNames = [...]string{kindGlobal: "global", kindChannel: "channel", kindRecipients: "recipients", kindDM: "dm"}
|
||||
simKindWeights = [...]int{kindGlobal: 30, kindChannel: 35, kindRecipients: 15, kindDM: 20}
|
||||
)
|
||||
|
||||
// simAlloc is one allocated seq and how the hub addressed it.
|
||||
type simAlloc struct {
|
||||
seq uint64
|
||||
ch int64
|
||||
kind simKind
|
||||
users []int64 // recipients or DM participants
|
||||
}
|
||||
|
||||
// reaches is the live audience rule for a registered client.
|
||||
func (a simAlloc) reaches(c *simClient) bool {
|
||||
switch a.kind {
|
||||
case kindGlobal:
|
||||
return true
|
||||
case kindChannel:
|
||||
return c.focus == a.ch
|
||||
default:
|
||||
return slices.Contains(a.users, c.user.ID)
|
||||
}
|
||||
}
|
||||
|
||||
type simClient struct {
|
||||
idx int
|
||||
user *db.User
|
||||
allowed map[int64]bool // READ-allowed channels: every text channel plus this user's DM channels
|
||||
conn *ws.Client // nil while disconnected
|
||||
send chan []byte
|
||||
high chan []byte
|
||||
low chan []byte
|
||||
wire *ws.FaultConn
|
||||
conns int
|
||||
cut bool // the transport cut the socket; the server has not noticed yet
|
||||
focus int64 // focused channel, 0 = none
|
||||
w uint64 // watermark: max seq read
|
||||
owed []uint64
|
||||
}
|
||||
|
||||
type sim struct {
|
||||
t *testing.T
|
||||
hub *ws.Hub
|
||||
rng *rand.Rand
|
||||
seed uint64
|
||||
steps int
|
||||
step int
|
||||
chIDs []int64
|
||||
clients []*simClient
|
||||
chanOf []int64 // chanOf[seq] = channel of every allocated seq; index 0 unused
|
||||
sched ws.FaultSchedule
|
||||
trace []string
|
||||
stats map[string]int // every figure here is a pure function of the seed
|
||||
racing int // racing seqs that landed inside a replay burst: the scheduler's call, not the seed's
|
||||
raced int // resumes where a broadcast allocated a seq while registration was still in flight: also the scheduler's
|
||||
}
|
||||
|
||||
func TestHubSimulation(t *testing.T) {
|
||||
seeds, steps := simConfig(t)
|
||||
total := map[string]int{}
|
||||
for _, seed := range seeds {
|
||||
t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) {
|
||||
stats, raced := runHubSim(t, seed, steps)
|
||||
for k, v := range stats {
|
||||
total[k] += v
|
||||
}
|
||||
total["raced"] += raced
|
||||
})
|
||||
}
|
||||
// The floor: every load-bearing transition must have happened at least
|
||||
// once across the run, or a constant change (weights, ring, queue, wire
|
||||
// schedule) could turn the simulation into no-ops with CI still green.
|
||||
// Every key but one is a function of the seed. "raced" — a broadcast
|
||||
// that allocated a seq while a resume's registration was still in flight
|
||||
// — is the scheduler's, so it is floored only in aggregate: the default
|
||||
// run has 179 resumes with a burst ("bursts", seed-determined), and each
|
||||
// one overlaps with odds far above a coin flip — registerNow makes two
|
||||
// slog syscalls before the goroutine can return, and 178–179 of the 179
|
||||
// were observed overlapping in three measured runs — so an all-miss run
|
||||
// is not luck but a scheduler or lock change that no longer lets a
|
||||
// broadcast run against a registration, the very thing this simulation
|
||||
// exists to exercise. Calibrated for the default seed count; a
|
||||
// single-seed replay or a shorter seed list skips it.
|
||||
if t.Failed() {
|
||||
return
|
||||
}
|
||||
if os.Getenv("OWNCORD_SIM_SEED") != "" || len(seeds) < simDefaultSeeds {
|
||||
t.Logf("hub simulation: floor skipped — OWNCORD_SIM_SEED=%q, %d seed(s) (calibrated for %d)", os.Getenv("OWNCORD_SIM_SEED"), len(seeds), simDefaultSeeds)
|
||||
return
|
||||
}
|
||||
t.Logf("hub simulation: floor totals %v", total)
|
||||
for _, k := range []string{"global", "channel", "recipients", "dm", "resume", "bursts", "raced", "fallback", "fresh", "cut", "kicked"} {
|
||||
if total[k] == 0 {
|
||||
t.Errorf("hub simulation: %q never happened across %d seeds x %d steps — the step mix no longer reaches it", k, len(seeds), steps)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func simConfig(t *testing.T) (seeds []uint64, steps int) {
|
||||
t.Helper()
|
||||
steps = simEnvInt(t, "OWNCORD_SIM_STEPS", simDefaultSteps)
|
||||
if v := os.Getenv("OWNCORD_SIM_SEED"); v != "" {
|
||||
seed, err := strconv.ParseUint(v, 10, 64)
|
||||
if err != nil {
|
||||
t.Fatalf("OWNCORD_SIM_SEED=%q: %v", v, err)
|
||||
}
|
||||
return []uint64{seed}, steps
|
||||
}
|
||||
for i := range simEnvInt(t, "OWNCORD_SIM_SEEDS", simDefaultSeeds) {
|
||||
seeds = append(seeds, uint64(i+1))
|
||||
}
|
||||
return seeds, steps
|
||||
}
|
||||
|
||||
func simEnvInt(t *testing.T, name string, def int) int {
|
||||
t.Helper()
|
||||
v := os.Getenv(name)
|
||||
if v == "" {
|
||||
return def
|
||||
}
|
||||
n, err := strconv.Atoi(v)
|
||||
if err != nil || n <= 0 {
|
||||
t.Fatalf("%s=%q: want a positive integer", name, v)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func simDMChannel(i, j int) int64 {
|
||||
return 1000 + int64(min(i, j))*simClients + int64(max(i, j))
|
||||
}
|
||||
|
||||
func runHubSim(t *testing.T, seed uint64, steps int) (stats map[string]int, raced int) {
|
||||
hub, database := newTestHub(t)
|
||||
hub.ConfigureReplay(simRing, 0)
|
||||
hub.FreezeTopicLimiterForTest()
|
||||
s := &sim{
|
||||
t: t, hub: hub, seed: seed, steps: steps,
|
||||
rng: rand.New(rand.NewPCG(seed, seed^0xD1B54A32D192ED03)),
|
||||
chanOf: []int64{0},
|
||||
stats: map[string]int{},
|
||||
}
|
||||
for i := range simChannels {
|
||||
s.chIDs = append(s.chIDs, seedTestChannel(t, database, fmt.Sprintf("sim-%d", i)))
|
||||
}
|
||||
for i := range simClients {
|
||||
c := &simClient{idx: i, user: seedOwnerUser(t, database, fmt.Sprintf("sim-user-%d", i)), allowed: map[int64]bool{}}
|
||||
for _, ch := range s.chIDs {
|
||||
c.allowed[ch] = true
|
||||
}
|
||||
for j := range simClients {
|
||||
if j != i {
|
||||
c.allowed[simDMChannel(i, j)] = true
|
||||
}
|
||||
}
|
||||
s.clients = append(s.clients, c)
|
||||
}
|
||||
// The wire: an order-preserving lag and the one fault TCP really has, a cut.
|
||||
s.sched = ws.FaultSchedule{Delay: s.rng.IntN(3), Drop: 0.02, DropTail: true}
|
||||
for _, c := range s.clients {
|
||||
s.connect(c)
|
||||
}
|
||||
for s.step = 1; s.step <= steps; s.step++ {
|
||||
c := s.clients[s.rng.IntN(simClients)]
|
||||
switch simOp(s.pick(simOpWeights[:])) {
|
||||
case opSubscribe:
|
||||
s.subscribe(c)
|
||||
case opBroadcast:
|
||||
s.broadcast(nil)
|
||||
case opAck:
|
||||
s.ack(c)
|
||||
case opDisconnect:
|
||||
s.disconnect(c)
|
||||
case opReconnect:
|
||||
s.reconnect(c)
|
||||
}
|
||||
s.checkCounts()
|
||||
}
|
||||
for _, c := range s.clients {
|
||||
if c.conn != nil && !c.cut {
|
||||
s.read(c, -1)
|
||||
}
|
||||
}
|
||||
t.Logf("seed %d: %d steps, seq %d, %v", seed, steps, hub.SeqForTest(), s.stats)
|
||||
t.Logf("seed %d: %d resume(s) overlapped a broadcast (raced), %d racing seq(s) landed inside a replay burst — both scheduler-decided, kept off the stats line", seed, s.raced, s.racing)
|
||||
return s.stats, s.raced
|
||||
}
|
||||
|
||||
func (s *sim) pick(weights []int) int {
|
||||
total := 0
|
||||
for _, w := range weights {
|
||||
total += w
|
||||
}
|
||||
r := s.rng.IntN(total)
|
||||
for i, w := range weights {
|
||||
if r < w {
|
||||
return i
|
||||
}
|
||||
r -= w
|
||||
}
|
||||
return len(weights) - 1
|
||||
}
|
||||
|
||||
func (s *sim) note(format string, args ...any) {
|
||||
s.trace = append(s.trace, fmt.Sprintf("#%d ", s.step)+fmt.Sprintf(format, args...))
|
||||
if len(s.trace) > 24 {
|
||||
s.trace = s.trace[1:]
|
||||
}
|
||||
}
|
||||
|
||||
func (s *sim) failf(format string, args ...any) {
|
||||
s.t.Helper()
|
||||
s.t.Fatalf("hub simulation: seed %d step %d: %s\nreplay: OWNCORD_SIM_SEED=%d OWNCORD_SIM_STEPS=%d go test -race -count=1 -run '^TestHubSimulation$' ./ws/\nlast steps:\n %s",
|
||||
s.seed, s.step, fmt.Sprintf(format, args...), s.seed, s.steps, strings.Join(s.trace, "\n "))
|
||||
}
|
||||
|
||||
// ─── connections ─────────────────────────────────────────────────────────────
|
||||
|
||||
func (s *sim) newConn(c *simClient, channelID int64, lastSeq uint64) *ws.Client {
|
||||
c.send = make(chan []byte, simSendBuf)
|
||||
c.high = make(chan []byte, 64)
|
||||
c.low = make(chan []byte, 64)
|
||||
return ws.NewSimClientForTest(s.hub, c.user, channelID, lastSeq, c.send, c.high, c.low)
|
||||
}
|
||||
|
||||
func (s *sim) attach(c *simClient, conn *ws.Client, preface [][]byte, owed []uint64) {
|
||||
c.conns++
|
||||
c.conn, c.cut, c.owed = conn, false, owed
|
||||
c.wire = ws.NewFaultConnForTest(s.seed, uint64(c.idx)<<32|uint64(c.conns), s.sched, preface, c.send)
|
||||
// Take what the queue already holds — the racing broadcasts that landed
|
||||
// after registerNow — into the wire now: the queue's fill, and so the
|
||||
// step at which it could overflow, must not depend on which side of the
|
||||
// snapshot those frames fell, the one thing the scheduler rather than
|
||||
// the seed decides.
|
||||
c.wire.Pull()
|
||||
if got := s.hub.GetClient(c.user.ID); got != conn {
|
||||
s.failf("c%d: hub holds %p for the user after registration, want %p", c.idx, got, conn)
|
||||
}
|
||||
}
|
||||
|
||||
// connect is a fresh connect (last_seq 0): handleFreshConnect registers with
|
||||
// a nil readable set, so nothing is inherited and there is no replay.
|
||||
func (s *sim) connect(c *simClient) {
|
||||
conn := s.newConn(c, 0, 0)
|
||||
s.hub.RegisterNowForTest(conn)
|
||||
c.focus = 0
|
||||
s.attach(c, conn, nil, nil)
|
||||
}
|
||||
|
||||
// teardownOld is the replaced socket's readPump defer finishing late (I6).
|
||||
func (s *sim) teardownOld(c *simClient, old *ws.Client) {
|
||||
if old != nil && !s.hub.UnregisterNowForTest(old) {
|
||||
s.failf("I6: c%d: the replaced connection's teardown evicted the replacement", c.idx)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── steps ───────────────────────────────────────────────────────────────────
|
||||
|
||||
func (s *sim) subscribe(c *simClient) {
|
||||
// A queue the server already closed (overflow kick, read as FaultClosed
|
||||
// at the next ack) is a dying socket: Subscribe refuses it by design, and
|
||||
// the client is not sending channel_focus on it anyway.
|
||||
if c.conn == nil || c.cut || ws.IsSendClosedForTest(c.conn) {
|
||||
s.note("subscribe c%d: no live connection", c.idx)
|
||||
return
|
||||
}
|
||||
var ch int64
|
||||
if s.rng.IntN(4) != 0 {
|
||||
ch = s.chIDs[s.rng.IntN(simChannels)]
|
||||
}
|
||||
s.hub.ApplySetChannelIDForTest(c.conn, ch)
|
||||
c.focus = ch
|
||||
if got := ws.ClientChannelIDForTest(c.conn); got != ch || (ch != 0 && !s.hub.SubscribedToChannelTopicForTest(c.conn, ch)) {
|
||||
s.failf("channel_focus %d left c%d focused on %d", ch, c.idx, got)
|
||||
}
|
||||
s.note("subscribe c%d -> ch%d", c.idx, ch)
|
||||
}
|
||||
|
||||
// broadcast allocates one seq through the real delivery path and records who
|
||||
// is owed it. racing, when set, is the client mid-reconnect: the caller
|
||||
// resolves its audience itself, and the frame is aimed so that it reaches
|
||||
// that client whether it lands before the snapshot (replayed) or after it
|
||||
// (live) — global, the channel it is focused on, a recipient list it is on,
|
||||
// a DM it is in. A frame the replay filter would carry but the live audience
|
||||
// would not (another channel, someone else's recipients) would make what the
|
||||
// client reads depend on the interleaving, and with it every later wire
|
||||
// draw; the replay-superset case is covered by every non-racing broadcast.
|
||||
func (s *sim) broadcast(racing *simClient) simAlloc {
|
||||
kind := simKind(s.pick(simKindWeights[:]))
|
||||
if racing != nil && kind == kindChannel && racing.focus == 0 {
|
||||
kind = kindGlobal
|
||||
}
|
||||
a := simAlloc{kind: kind}
|
||||
payload := fmt.Appendf(nil, `{"type":"sim","step":%d}`, s.step)
|
||||
before := s.hub.SeqForTest()
|
||||
switch a.kind {
|
||||
case kindGlobal:
|
||||
a.seq = s.hub.DeliverBroadcastForTest(0, nil, payload)
|
||||
case kindChannel:
|
||||
a.ch = s.chIDs[s.rng.IntN(simChannels)]
|
||||
if racing != nil {
|
||||
a.ch = racing.focus
|
||||
}
|
||||
a.seq = s.hub.DeliverBroadcastForTest(a.ch, nil, payload)
|
||||
case kindRecipients:
|
||||
a.ch = s.chIDs[s.rng.IntN(simChannels)]
|
||||
a.users = make([]int64, 0, simClients)
|
||||
for _, c := range s.clients {
|
||||
if c == racing || s.rng.IntN(2) == 0 {
|
||||
a.users = append(a.users, c.user.ID)
|
||||
}
|
||||
}
|
||||
a.seq = s.hub.DeliverBroadcastForTest(a.ch, a.users, payload)
|
||||
case kindDM:
|
||||
i := s.rng.IntN(simClients)
|
||||
if racing != nil {
|
||||
i = racing.idx
|
||||
}
|
||||
j := (i + 1 + s.rng.IntN(simClients-1)) % simClients
|
||||
a.ch = simDMChannel(i, j)
|
||||
a.users = []int64{s.clients[i].user.ID, s.clients[j].user.ID}
|
||||
a.seq = s.hub.SendSequencedToUsersForTest(a.ch, a.users, payload)
|
||||
}
|
||||
after := s.hub.SeqForTest()
|
||||
switch {
|
||||
case a.seq == 0 && after != before:
|
||||
s.failf("I4: a shed %s frame advanced seq %d -> %d", simKindNames[a.kind], before, after)
|
||||
case a.seq == 0:
|
||||
s.stats["shed"]++
|
||||
s.note("broadcast %s ch%d: shed by the topic limiter", simKindNames[a.kind], a.ch)
|
||||
return a
|
||||
case after != a.seq || s.hub.ReplayBuffer().NewestSeq() != a.seq:
|
||||
s.failf("I4: seq %d allocated but ring newest is %d (hub seq %d)", a.seq, s.hub.ReplayBuffer().NewestSeq(), after)
|
||||
case uint64(len(s.chanOf)) != a.seq:
|
||||
s.failf("I4: seq %d is not the successor of %d", a.seq, len(s.chanOf)-1)
|
||||
}
|
||||
s.chanOf = append(s.chanOf, a.ch)
|
||||
for _, c := range s.clients {
|
||||
if c != racing && c.conn != nil && !c.cut && a.reaches(c) {
|
||||
c.owed = append(c.owed, a.seq)
|
||||
}
|
||||
}
|
||||
s.stats[simKindNames[a.kind]]++
|
||||
s.note("broadcast %s ch%d users%v -> seq %d", simKindNames[a.kind], a.ch, a.users, a.seq)
|
||||
return a
|
||||
}
|
||||
|
||||
func (s *sim) ack(c *simClient) {
|
||||
if c.conn == nil || c.cut {
|
||||
s.note("ack c%d: no live connection", c.idx)
|
||||
return
|
||||
}
|
||||
n := 1 + s.rng.IntN(8)
|
||||
if s.rng.IntN(2) == 0 {
|
||||
n = -1 // drain
|
||||
}
|
||||
s.read(c, n)
|
||||
}
|
||||
|
||||
// read pulls up to n frames (n < 0: until the wire is empty) through the
|
||||
// transport and checks each against the model.
|
||||
func (s *sim) read(c *simClient, n int) {
|
||||
defer s.checkPriorityQueues(c)
|
||||
read := 0
|
||||
for n < 0 || read < n {
|
||||
frame, st := c.wire.Recv()
|
||||
if st == ws.FaultOK {
|
||||
s.observe(c, frame)
|
||||
read++
|
||||
continue
|
||||
}
|
||||
switch st {
|
||||
case ws.FaultEmpty:
|
||||
if len(c.owed) != c.wire.Buffered() {
|
||||
s.failf("I2: c%d conn %d: owed %v but only %d unread frame(s) remain (W=%d)", c.idx, c.conns, c.owed, c.wire.Buffered(), c.w)
|
||||
}
|
||||
s.note("ack c%d: read %d, W=%d, %d in flight", c.idx, read, c.w, c.wire.Buffered())
|
||||
case ws.FaultClosed:
|
||||
// The server closed the queue: the normal-queue overflow kick
|
||||
// (client.go sendMsg, BUG-124). Its readPump defer follows.
|
||||
if !ws.IsSendClosedForTest(c.conn) || s.hub.UnregisterNowForTest(c.conn) {
|
||||
s.failf("I6: c%d: queue reported closed on a live connection, or its teardown reported replaced", c.idx)
|
||||
}
|
||||
c.conn, c.owed = nil, nil
|
||||
s.stats["kicked"]++
|
||||
s.note("ack c%d: read %d then the server had closed the queue (overflow kick), W=%d", c.idx, read, c.w)
|
||||
case ws.FaultCut:
|
||||
c.cut, c.owed = true, nil
|
||||
s.stats["cut"]++
|
||||
s.note("ack c%d: read %d then the wire cut, W=%d", c.idx, read, c.w)
|
||||
}
|
||||
return
|
||||
}
|
||||
s.note("ack c%d: read %d, W=%d", c.idx, read, c.w)
|
||||
}
|
||||
|
||||
func (s *sim) disconnect(c *simClient) {
|
||||
if c.conn == nil {
|
||||
s.note("disconnect c%d: already gone", c.idx)
|
||||
return
|
||||
}
|
||||
if s.hub.UnregisterNowForTest(c.conn) || s.hub.GetClient(c.user.ID) != nil {
|
||||
s.failf("I6: c%d: tearing down the live connection reported replaced or left a client registered", c.idx)
|
||||
}
|
||||
ws.CloseSendForTest(c.conn)
|
||||
c.conn, c.owed, c.cut = nil, nil, false
|
||||
s.note("disconnect c%d (W=%d)", c.idx, c.w)
|
||||
}
|
||||
|
||||
// reconnect is the resume: a new socket for the same user carrying W as
|
||||
// last_seq, registered through reconnectRegister — the replay snapshot and
|
||||
// registerNow under one h.seqMu section — while up to three broadcasts race
|
||||
// it from this goroutine, exactly the interleaving the seqMu discipline
|
||||
// exists for. The old socket may still be registered (a network blip: the
|
||||
// transfer path) or already gone (active_channel_id restores the focus).
|
||||
func (s *sim) reconnect(c *simClient) {
|
||||
old := c.conn
|
||||
if c.w == 0 {
|
||||
s.connect(c)
|
||||
s.teardownOld(c, old)
|
||||
s.stats["fresh"]++
|
||||
s.note("reconnect c%d: fresh (W=0)", c.idx)
|
||||
return
|
||||
}
|
||||
// The auth frame's active_channel_id: none, the channel the client had
|
||||
// open, or one it switched to while offline. handleReconnect honours it
|
||||
// when READ-allowed (every text channel here), and registerNow lets it
|
||||
// win over the transfer, which lands only on a client that declared none.
|
||||
var authCh int64
|
||||
switch s.rng.IntN(3) {
|
||||
case 1:
|
||||
authCh = c.focus
|
||||
case 2:
|
||||
authCh = s.chIDs[s.rng.IntN(simChannels)]
|
||||
}
|
||||
if authCh != 0 || old == nil {
|
||||
c.focus = authCh
|
||||
}
|
||||
conn := s.newConn(c, authCh, c.w)
|
||||
s0 := s.hub.SeqForTest()
|
||||
var (
|
||||
events [][]byte
|
||||
ok bool
|
||||
done = make(chan struct{})
|
||||
)
|
||||
go func() {
|
||||
defer close(done)
|
||||
events, ok = s.hub.ReconnectRegisterForTest(conn, c.w, c.allowed)
|
||||
}()
|
||||
burst := s.rng.IntN(4)
|
||||
// Racing allocations can evict W from the ring before or after the
|
||||
// snapshot. Where that would decide replay-or-fallback, the scheduler
|
||||
// would be choosing the path and everything the client reads after it,
|
||||
// so a resume that close to the eviction boundary is not raced; the
|
||||
// paths on either side of the boundary are the seed's alone. What stays
|
||||
// the scheduler's is only the split of the racing seqs between the
|
||||
// replay burst and the live queue.
|
||||
if oldest := s.hub.ReplayBuffer().OldestSeq(); oldest < c.w && c.w <= oldest+uint64(burst) {
|
||||
burst = 0
|
||||
}
|
||||
var racing []simAlloc
|
||||
overlapped := false
|
||||
for range burst {
|
||||
a := s.broadcast(c)
|
||||
if a.seq == 0 {
|
||||
continue
|
||||
}
|
||||
racing = append(racing, a)
|
||||
// An overlap on record: this seq was allocated while the goroutine
|
||||
// had not yet been seen to return, so the broadcast ran against a
|
||||
// registration in flight. A burst the goroutine beat to the lock, or
|
||||
// one the limiter shed entirely, is not one.
|
||||
select {
|
||||
case <-done:
|
||||
default:
|
||||
overlapped = true
|
||||
}
|
||||
}
|
||||
<-done
|
||||
var owed []uint64
|
||||
if ok {
|
||||
owed = s.checkReplay(c, events, s0, racing)
|
||||
s.stats["resume"]++
|
||||
if burst > 0 {
|
||||
s.stats["bursts"]++
|
||||
}
|
||||
if overlapped {
|
||||
s.raced++
|
||||
}
|
||||
} else {
|
||||
if s.hub.ReplayBuffer().EventsSinceFiltered(c.w, c.allowed) != nil {
|
||||
s.failf("I5: c%d: replay refused although the ring (oldest %d) covers W=%d", c.idx, s.hub.ReplayBuffer().OldestSeq(), c.w)
|
||||
}
|
||||
// handleFreshConnect's replay-failure fallback: registerNow with the
|
||||
// readable set, so the focus still transfers; a full ready follows.
|
||||
s.hub.RegisterNowWithReadableForTest(conn, c.allowed)
|
||||
events = nil
|
||||
s.stats["fallback"]++
|
||||
}
|
||||
s.teardownOld(c, old)
|
||||
s.attach(c, conn, events, owed)
|
||||
if got := ws.ClientChannelIDForTest(conn); got != c.focus {
|
||||
s.failf("c%d: resumed connection focused on %d, want %d", c.idx, got, c.focus)
|
||||
}
|
||||
s.note("reconnect c%d: W=%d old=%v auth_ch=%d replay=%v racing=%d owed=%v", c.idx, c.w, old != nil, authCh, ok, len(racing), owed)
|
||||
}
|
||||
@@ -17,7 +17,7 @@ import (
|
||||
|
||||
// ─── test helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
func openTestDB(t *testing.T) *db.DB {
|
||||
func openTestDB(t testing.TB) *db.DB {
|
||||
t.Helper()
|
||||
database, err := db.Open(":memory:")
|
||||
if err != nil {
|
||||
@@ -34,7 +34,7 @@ func openTestDB(t *testing.T) *db.DB {
|
||||
return database
|
||||
}
|
||||
|
||||
func newTestHub(t *testing.T) (*ws.Hub, *db.DB) {
|
||||
func newTestHub(t testing.TB) (*ws.Hub, *db.DB) {
|
||||
t.Helper()
|
||||
database := openTestDB(t)
|
||||
limiter := auth.NewRateLimiter()
|
||||
@@ -43,7 +43,7 @@ func newTestHub(t *testing.T) (*ws.Hub, *db.DB) {
|
||||
}
|
||||
|
||||
// seedTestUser inserts a Member-role user and returns its ID.
|
||||
func seedTestUser(t *testing.T, database *db.DB, username string) int64 {
|
||||
func seedTestUser(t testing.TB, database *db.DB, username string) int64 {
|
||||
t.Helper()
|
||||
id, err := database.CreateUser(context.Background(), username, "hash", 4)
|
||||
if err != nil {
|
||||
@@ -54,7 +54,7 @@ func seedTestUser(t *testing.T, database *db.DB, username string) int64 {
|
||||
|
||||
// seedOwnerUser inserts an Owner-role user and returns the full *db.User.
|
||||
// Owner role (id=1) has all permissions (0x7FFFFFFF), so it passes all checks.
|
||||
func seedOwnerUser(t *testing.T, database *db.DB, username string) *db.User {
|
||||
func seedOwnerUser(t testing.TB, database *db.DB, username string) *db.User {
|
||||
t.Helper()
|
||||
_, err := database.CreateUser(context.Background(), username, "hash", 1) // roleID=1 → Owner
|
||||
if err != nil {
|
||||
@@ -68,7 +68,7 @@ func seedOwnerUser(t *testing.T, database *db.DB, username string) *db.User {
|
||||
}
|
||||
|
||||
// seedTestChannel inserts a channel and returns its ID.
|
||||
func seedTestChannel(t *testing.T, database *db.DB, name string) int64 {
|
||||
func seedTestChannel(t testing.TB, database *db.DB, name string) int64 {
|
||||
t.Helper()
|
||||
id, err := database.CreateChannel(context.Background(), name, "text", "", "", 0)
|
||||
if err != nil {
|
||||
|
||||
@@ -1077,6 +1077,121 @@ db`). No control file is committed.
|
||||
with no `timeout-minutes`, so it inherited GitHub's 360-minute default.
|
||||
Hence the one-line `ci.yml` change; the new workflow declares its own.
|
||||
|
||||
#### Evidence — items 2 + 3 (hub simulation + fault-injected transport)
|
||||
|
||||
- Branch `feat/b3-6-hub-sim`; commits: 3c5d75f8 `test(b3-6): seeded hub
|
||||
simulation and fault-injected transport (items 2 and 3)`, 932b5d6b the docs
|
||||
commit carrying this block, and the review-fix commit `fix(b3-6): hub sim
|
||||
— deterministic topic limiter for exact replay, a floor on the step mix,
|
||||
auth-frame-wins under transfer, wire-seed mixing` (it carries this text, so
|
||||
its own SHA is in the PR, not here).
|
||||
- Oracle (the doc comment of `Server/ws/hub_sim_test.go`, written before the
|
||||
driver): I1 per-connection strictly increasing `seq`, never on the high or
|
||||
low queue; I2 a live connection yields exactly the seqs allocated for its
|
||||
audience, in order — nothing missing, extra or twice — with the unread count
|
||||
checked after every step; I3 a resume from watermark W is replayed exactly
|
||||
`{s in (W, S] : channel 0 or READ-allowed}` where S is the hub seq at the
|
||||
instant `registerNow` ran (atomic with the snapshot under `seqMu`) and every
|
||||
audience seq above S arrives live; I4 `h.seq` advances only for a frame that
|
||||
reached the ring; I5 an evicted W is refused a replay and registered as the
|
||||
full-ready fallback; I6 a replaced socket's late teardown reports
|
||||
`replaced=true`. The one narrowing: frames the dying socket still held above
|
||||
W are delivered again by the replay — a cross-connection duplicate the
|
||||
max(seq) ack makes by design — and the oracle allows exactly that.
|
||||
- RED (a): inverted the I2 head check →
|
||||
`hub simulation: seed 1 step 28: I2: c3 conn 2: expected seq 4 next, got 4 (owed [4 6 8 9])`
|
||||
plus the replay line, on 20/20 seeds.
|
||||
- RED (b): `OWNCORD_SIM_SEED=1 OWNCORD_SIM_STEPS=200 go test -race -count=1 -run '^TestHubSimulation$' ./ws/`
|
||||
→ `seed 1 step 28: I2: c3 conn 2: expected seq 4 next, got 4` — the same
|
||||
step, the same frame.
|
||||
- RED (c): `FaultSchedule{Drop: 1}` (silent, no tail cut) on the resumed
|
||||
connection's wire →
|
||||
`seed 1 step 121: I2: c6 conn 3: owed [45 46 47 48 49] but only 0 unread frame(s) remain (W=44)`
|
||||
on 20/20 seeds. With `DropTail` instead, the same drop is a socket death and
|
||||
correctly invisible: the client resumes from W and the replay repairs it.
|
||||
- RED (d, extra): replay snapshot and `registerNow` outside one `seqMu`
|
||||
section with a 1 ms gap →
|
||||
`seed 1 step 100: I2: c3 conn 3 holds 33 unread frame(s) but is owed 34`
|
||||
on 19/20 seeds — the registerNow-gap class `hub_register_race_test.go` pins,
|
||||
found by the generated orderings rather than a hand-written interleaving.
|
||||
- GREEN: `go test -race -count=1 -run '^TestHubSimulation$' ./ws/` →
|
||||
`ok github.com/J3vb/OwnCord/Server/ws 2.258s`.
|
||||
- Exact replay (review fix): the topic limiter is frozen to a per-channel
|
||||
count (`FreezeTopicLimiterForTest`), the racing burst is aimed only at the
|
||||
resuming client's own audience, its frames are pulled into the wire at
|
||||
attach time, and a resume within the burst's reach of the ring's eviction
|
||||
boundary is not raced — so every figure on the stats line is a pure
|
||||
function of the seed. Proof, three runs of
|
||||
`OWNCORD_SIM_SEED=1 OWNCORD_SIM_STEPS=10000 go test -race -count=1 -v -run '^TestHubSimulation$' ./ws/`:
|
||||
`seed 1: 10000 steps, seq 4500, map[bursts:508 channel:300 cut:142 dm:1285 fallback:751 fresh:22 global:1975 kicked:38 recipients:940 resume:710 shed:1630]`
|
||||
three times, byte-identical. What still varies between runs is the
|
||||
scheduler's interleaving inside a racing reconnect step — how many of the
|
||||
racing seqs land in the replay burst rather than the live queue (420, 424
|
||||
and 425 in earlier runs) and how many resumes are observed overlapping a
|
||||
broadcast at all (`raced`), both printed on a second line — so a model
|
||||
defect that depends on that split replays only probabilistically; run the
|
||||
seed with `-count`.
|
||||
- Floor (review fix, Codex P2 on #1458): `TestHubSimulation` aggregates the
|
||||
per-seed stats and requires `global`, `channel`, `recipients`, `dm`,
|
||||
`resume`, `bursts` (a resume that got a replay with a burst requested),
|
||||
`fallback`, `fresh`, `cut` and `kicked` each ≥ 1 across the default run —
|
||||
every one a function of the seed — plus `raced`, a resume where a
|
||||
broadcast allocated a seq while the registration goroutine had not yet
|
||||
been observed to return. That overlap is the scheduler's, so `raced` lives
|
||||
on the second (scheduler-decided) line, not the stats line, and is floored
|
||||
only in aggregate across the 20 seeds: the default run has 179 bursts and
|
||||
178–179 of them were observed overlapping in three measured runs
|
||||
(registerNow makes two slog syscalls before the goroutine can return), so
|
||||
an all-miss run is a scheduler or lock change, not luck. Skipped, with a
|
||||
log line, for `OWNCORD_SIM_SEED` or fewer than 20 seeds; totals logged.
|
||||
RED: reconnect weight set to 0 →
|
||||
`hub simulation: "resume" never happened across 20 seeds x 200 steps — the step mix no longer reaches it`
|
||||
(and `bursts`, `fallback`, `fresh`); goroutine joined before the burst →
|
||||
`hub simulation: "raced" never happened across 20 seeds x 200 steps — the step mix no longer reaches it`;
|
||||
both restored.
|
||||
- Numbers: 8 clients, 3 channels, ring 48, normal queue 12; default 20 seeds ×
|
||||
200 steps in 2.3 s under `-race` (CI budget: under 10 s); `make sim` = 20 ×
|
||||
10,000 steps in 16.5 s (18 s wall with compile) under `-race`; per seed at 200 steps ≈ 13
|
||||
buffer resumes, 5 evicted-ring fallbacks, 12 fresh reconnects, 4 wire cuts,
|
||||
1–3 overflow kicks on about half the seeds, ~105 seqs; at 10,000 steps per
|
||||
seed ≈ 4,500 seqs, 700 resumes, 750 fallbacks, 140 cuts, 40 kicks, 300
|
||||
channel frames allocated and ≈1,630 shed by the frozen limiter;
|
||||
`BenchmarkReconnectStorm-32` 1590 ops, 702,410 ns/op for 50 resumes (≈14 µs
|
||||
each), 608,031 B/op, 954 allocs/op. Gates on the fix commit: `go vet ./...`
|
||||
clean; `go test -race ./ws/` → `ok 119.767s`;
|
||||
`go test -tags deadlock -count=10 -timeout 60m ./ws/` → `ok 586.646s`;
|
||||
`golangci-lint run` → 0 issues; `npm run check:docs` passed; prettier
|
||||
clean. (Before the fix: the four build variants clean, deadlock ×10
|
||||
`ok 594.098s`, `go test -race ./...` every package ok.)
|
||||
- Not simulated (the driver's five steps only): hub-originated frames —
|
||||
`hub.Run()` is never started, the driver is the sole allocator; tier
|
||||
selection is always "buffer", so the db tier, `mustFullResync` and
|
||||
visibility resyncs never interleave with a resume; `c.allowed` is the
|
||||
driver's own map (every text channel plus synthetic DM ids), not
|
||||
`computeAllowedChannels`, so permission variance is out of scope
|
||||
(`hub_register_test.go` covers the denial branch); there is no `writePump`,
|
||||
so I1's high/low-queue check is vacuous today — nothing in the step mix
|
||||
emits a priority frame; no voice supersession, no concurrent allocators.
|
||||
- Epoch harness: no use added. `TestEpoch1Fixtures` reads real sockets with
|
||||
expects interleaved across two connections, so a lag would wait for frames
|
||||
the journey has not produced yet (a deadlock) and any drop, duplicate or
|
||||
reorder changes the frame list the fixture pins; only the identity schedule
|
||||
is harmless, and that proves nothing. `NewFaultConnForTest` stays exported
|
||||
for item 4's client model and for a network-pattern proxy if one is needed.
|
||||
- Verified against HEAD: there is no ack message — the only ack is `last_seq`
|
||||
on the next auth frame — so the simulation's ack step is the client reading
|
||||
frames and advancing max(seq); the headless pattern expresses
|
||||
reconnect-transfer faithfully because `reconnectRegister` (the snapshot and
|
||||
`registerNow` under one `seqMu` section) is exported as-is, not
|
||||
re-implemented, and the fresh/fallback paths call the same `registerNow`
|
||||
handleFreshConnect does; no production file changed
|
||||
(`newTestHub`/`openTestDB`/`seedOwnerUser`/`seedTestChannel`/`seedTestUser`
|
||||
in `hub_test.go` now take `testing.TB` so the benchmark can share them). The
|
||||
shared rules' `go test -tags deadlock -count=10 ./ws/` overruns `go test`'s
|
||||
default 10-minute timeout on a 16-core desktop (601 s, "test timed out
|
||||
after 10m0s" with a goroutine dump that is not a detector hit); the gate
|
||||
needs `-timeout 60m`.
|
||||
|
||||
## B3-7 — Alpha-shaped test dataset
|
||||
|
||||
Roadmap workstream 12. Beside the slice.
|
||||
@@ -1314,5 +1429,9 @@ appended to the HP-3 scorecard as a dated "B3 exit" section the owner signs.
|
||||
- **Check the PR is still open before pushing a review fix** (HP-2 obs #97).
|
||||
- **`make` is not on PATH on Windows**; `npm run check:server` runs the same
|
||||
steps. `go test -tags deadlock -count=10 ./ws/` after every `ws` move.
|
||||
- **Ten deadlock passes overrun `go test`'s default 10-minute timeout** (B3-6
|
||||
items 2+3): write it as `go test -tags deadlock -count=10 -timeout 60m ./ws/`.
|
||||
A "test timed out after 10m0s" goroutine dump is the timeout, not a
|
||||
detector hit.
|
||||
- **`check:docs` counts.** `docs/plans/README.md` is watched; the register's
|
||||
row count and the ledger's status counts must agree with it.
|
||||
|
||||
Reference in New Issue
Block a user