| 1 | package eventwire |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "strings" |
| 6 | "testing" |
| 7 | ) |
| 8 | |
| 9 | func TestMirrorQueueBoundsDeltasAndRetainsLifecycleTruth(t *testing.T) { |
| 10 | var q MirrorQueue |
| 11 | for i := range MirrorQueueMaxFrames { |
| 12 | q.Push(Event{Kind: "text", Sequence: uint64(i + 1), Text: "delta"}) |
| 13 | } |
| 14 | if got, want := q.Len(), MirrorQueueMaxFrames-MirrorQueuePriorityReserve; got != want { |
| 15 | t.Fatalf("delta queue len = %d, want %d", got, want) |
| 16 | } |
| 17 | for i := range MirrorQueuePriorityReserve { |
| 18 | if !q.Push(Event{Kind: "notice", Code: "ordinary", Sequence: uint64(i + 1)}) { |
| 19 | t.Fatalf("priority frame %d was not admitted", i) |
| 20 | } |
| 21 | } |
| 22 | if got := q.Len(); got != MirrorQueueMaxFrames { |
| 23 | t.Fatalf("full queue len = %d", got) |
| 24 | } |
| 25 | if !q.Push(Event{Kind: "turn_done", TurnID: "latest"}) { |
| 26 | t.Fatal("must-reach turn_done was dropped") |
| 27 | } |
| 28 | frames := q.Take(MirrorQueueMaxFrames) |
| 29 | if got := frames[len(frames)-1]; got.Kind != "turn_done" || got.TurnID != "latest" { |
| 30 | t.Fatalf("last frame = %+v, want latest turn_done", got) |
| 31 | } |
| 32 | } |
| 33 | |
| 34 | func TestMirrorQueueByteBoundAndFailedBatchOrdering(t *testing.T) { |
| 35 | var q MirrorQueue |
| 36 | chunk := strings.Repeat("x", 1<<20) |
| 37 | for i := range 32 { |
| 38 | q.Push(Event{Kind: "text", Sequence: uint64(i + 10), Text: chunk}) |
| 39 | } |
| 40 | if q.Bytes() > MirrorQueueMaxBytes || q.Len() == 0 { |
| 41 | t.Fatalf("queue = %d frames, %d bytes", q.Len(), q.Bytes()) |
| 42 | } |
| 43 | q.Prepend([]Event{{Kind: "text", Sequence: 1}, {Kind: "text", Sequence: 2}}) |
| 44 | frames := q.Take(2) |
| 45 | if len(frames) != 2 || frames[0].Sequence != 1 || frames[1].Sequence != 2 { |
| 46 | t.Fatalf("retry order = %+v", frames) |
| 47 | } |
| 48 | } |
| 49 | |
| 50 | func TestMirrorQueueRetainsNewestOwnershipNoticeAtSaturation(t *testing.T) { |
| 51 | var q MirrorQueue |
| 52 | for i := range MirrorQueueMaxFrames { |
| 53 | q.Push(Event{Kind: "turn_done", TurnID: strings.Repeat("x", 8), Sequence: uint64(i + 1)}) |
| 54 | } |
| 55 | if !q.Push(Event{Kind: "notice", Code: "session_reclaim_requested", Sequence: 9999}) { |
| 56 | t.Fatal("ownership notice was dropped") |
| 57 | } |
| 58 | frames := q.Take(MirrorQueueMaxFrames) |
| 59 | last := frames[len(frames)-1] |
| 60 | if last.Code != "session_reclaim_requested" || last.Sequence != 9999 { |
| 61 | t.Fatalf("last frame = %+v", last) |
| 62 | } |
| 63 | } |
| 64 | |
| 65 | func TestMarshalMirrorBatchUsesActualEnvelopeSize(t *testing.T) { |
| 66 | frames := []Event{ |
| 67 | {Kind: "text", Text: strings.Repeat("a", 800)}, |
| 68 | {Kind: "text", Text: strings.Repeat("b", 800)}, |
| 69 | } |
| 70 | marshal := func(batch []Event) ([]byte, error) { |
| 71 | return json.Marshal(map[string]any{ |
| 72 | "sessionPath": strings.Repeat("p", 80), |
| 73 | "mirrorId": strings.Repeat("m", 80), |
| 74 | "frames": batch, |
| 75 | }) |
| 76 | } |
| 77 | one, remainder, payload, err := MarshalMirrorBatch(frames, 1200, marshal) |
| 78 | if err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | if len(one) != 1 || len(remainder) != 1 || len(payload) > 1200 { |
| 82 | t.Fatalf("batch=%d remainder=%d bytes=%d", len(one), len(remainder), len(payload)) |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | func TestMarshalMirrorBatchRejectsOversizedFirstFrameInTwoMarshals(t *testing.T) { |
| 87 | frames := make([]Event, MirrorBatchMaxFrames) |
| 88 | for i := range frames { |
| 89 | frames[i] = Event{Kind: "text", Sequence: uint64(i + 1)} |
| 90 | } |
| 91 | calls := 0 |
| 92 | marshal := func(batch []Event) ([]byte, error) { |
| 93 | calls++ |
| 94 | if len(batch) == 0 { |
| 95 | return []byte(`{"frames":[]}`), nil |
| 96 | } |
| 97 | return make([]byte, MirrorBatchMaxBytes+1), nil |
| 98 | } |
| 99 | batch, remainder, payload, err := MarshalMirrorBatch(frames, MirrorBatchMaxBytes, marshal) |
| 100 | if err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | if calls != 2 { |
| 104 | t.Fatalf("marshal calls = %d, want 2", calls) |
| 105 | } |
| 106 | if len(batch) != 0 || len(remainder) != len(frames) || len(payload) > MirrorBatchMaxBytes { |
| 107 | t.Fatalf("batch=%d remainder=%d payload=%d", len(batch), len(remainder), len(payload)) |
| 108 | } |
| 109 | for i, frame := range remainder { |
| 110 | if frame.Sequence != uint64(i+1) { |
| 111 | t.Fatalf("remainder[%d].sequence = %d", i, frame.Sequence) |
| 112 | } |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | func TestMarshalMirrorBatchBinarySearchesLargestOrderedPrefix(t *testing.T) { |
| 117 | frames := make([]Event, MirrorBatchMaxFrames) |
| 118 | for i := range frames { |
| 119 | frames[i] = Event{Kind: "text", Sequence: uint64(i + 1)} |
| 120 | } |
| 121 | const ( |
| 122 | envelopeBytes = 37 |
| 123 | frameBytes = 101 |
| 124 | wantFrames = 173 |
| 125 | ) |
| 126 | maxBytes := envelopeBytes + wantFrames*frameBytes |
| 127 | calls := 0 |
| 128 | marshal := func(batch []Event) ([]byte, error) { |
| 129 | calls++ |
| 130 | return make([]byte, envelopeBytes+len(batch)*frameBytes), nil |
| 131 | } |
| 132 | batch, remainder, payload, err := MarshalMirrorBatch(frames, maxBytes, marshal) |
| 133 | if err != nil { |
| 134 | t.Fatal(err) |
| 135 | } |
| 136 | if calls > 11 { |
| 137 | t.Fatalf("marshal calls = %d, want at most 11", calls) |
| 138 | } |
| 139 | if len(batch) != wantFrames || len(remainder) != len(frames)-wantFrames || len(payload) != maxBytes { |
| 140 | t.Fatalf("batch=%d remainder=%d payload=%d calls=%d", len(batch), len(remainder), len(payload), calls) |
| 141 | } |
| 142 | for i, frame := range batch { |
| 143 | if frame.Sequence != uint64(i+1) { |
| 144 | t.Fatalf("batch[%d].sequence = %d", i, frame.Sequence) |
| 145 | } |
| 146 | } |
| 147 | for i, frame := range remainder { |
| 148 | if frame.Sequence != uint64(wantFrames+i+1) { |
| 149 | t.Fatalf("remainder[%d].sequence = %d", i, frame.Sequence) |
| 150 | } |
| 151 | } |
| 152 | } |
| 153 |