1package presented
2
3import (
4 "context"
5 "reflect"
6 "testing"
7
8 "github.com/bnema/neferwl/internal/ports"
9 "github.com/stretchr/testify/require"
10)
11
12func ptr(m map[ports.WindowID]uint64) uintptr { return reflect.ValueOf(m).Pointer() }
13
14// An unchanged publication reuses its immutable snapshots; a change clones,
15// and earlier reports keep their maps.
16func TestQueueSnapshotsCloneOnlyOnChange(t *testing.T) {
17 var q Queue
18 seen := map[ports.WindowID]uint64{1: 4, 5: 7}
19 reads := map[ports.WindowID]uint64{5: 2}
20 q.Push(ports.OutputPresented{Output: "A", Seen: seen, ChildReads: reads})
21 first := q.Unsent()[0]
22 require.Equal(t, seen, first.Seen)
23 require.Equal(t, reads, first.ChildReads)
24 seen[1] = 99 // the caller's scratch changes: the report does not
25 require.Equal(t, uint64(4), first.Seen[1])
26
27 seen[1] = 4
28 q.Push(ports.OutputPresented{Output: "A", Seen: seen, ChildReads: reads})
29 again := q.Unsent()[0]
30 require.Equal(t, ptr(first.Seen), ptr(again.Seen), "unchanged seen is shared")
31 require.Equal(t, ptr(first.ChildReads), ptr(again.ChildReads), "unchanged reads are shared")
32 allocs := testing.AllocsPerRun(100, func() { q.Push(ports.OutputPresented{Output: "A", Seen: seen, ChildReads: reads}) })
33 require.Zero(t, allocs, "unchanged publication allocates nothing")
34
35 reads[5] = 3
36 q.Push(ports.OutputPresented{Output: "A", Seen: seen, ChildReads: reads})
37 changed := q.Unsent()[0]
38 require.Equal(t, map[ports.WindowID]uint64{5: 3}, changed.ChildReads)
39 require.Equal(t, map[ports.WindowID]uint64{5: 2}, first.ChildReads, "sent snapshot stays immutable")
40 require.Equal(t, ptr(first.Seen), ptr(changed.Seen), "seen still unchanged")
41}
42
43// Flip-less reports fold into the newest one and keep its flip; flips queue.
44func TestQueueKeepsEveryFlip(t *testing.T) {
45 var q Queue
46 seen := map[ports.WindowID]uint64{1: 1}
47 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 1}, Seen: seen})
48 q.Push(ports.OutputPresented{Seen: seen})
49 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 2}, Seen: seen})
50 q.Push(ports.OutputPresented{Seen: map[ports.WindowID]uint64{1: 3}})
51 u := q.Unsent()
52 require.Len(t, u, 2)
53 require.Equal(t, uint64(1), u[0].Flip.Seq)
54 require.Equal(t, uint64(2), u[1].Flip.Seq, "a flip-less report keeps the queued flip")
55 require.Equal(t, uint64(3), u[1].Seen[1])
56}
57
58// Past MaxUnsent the oldest flip folds into the next: counted as merged and
59// its shown windows carried, without mutating a shared Shows snapshot.
60func TestQueueMergesOldestFlipPastBound(t *testing.T) {
61 var q Queue
62 shared := map[ports.WindowID]uint64{2: 5}
63 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 1, Shows: map[ports.WindowID]uint64{1: 4}}})
64 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 2, Shows: shared}})
65 merged := false
66 for i := range MaxUnsent - 1 {
67 merged = q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: uint64(10 + i)}}) || merged
68 }
69 require.True(t, merged)
70 u := q.Unsent()
71 require.Len(t, u, MaxUnsent)
72 require.Equal(t, uint64(2), u[0].Flip.Seq)
73 require.Equal(t, 1, u[0].Flip.Merged)
74 require.Equal(t, map[ports.WindowID]uint64{1: 4, 2: 5}, u[0].Flip.Shows)
75 require.Equal(t, map[ports.WindowID]uint64{2: 5}, shared, "shared snapshot untouched")
76}
77
78func TestQueueFlushStopsWhenFull(t *testing.T) {
79 var q Queue
80 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 1}})
81 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 2}})
82 ch := make(chan ports.OutputPresented, 1)
83 q.Flush(ch)
84 require.Equal(t, 1, q.Len())
85 require.Equal(t, uint64(1), (<-ch).Flip.Seq)
86 require.NoError(t, q.Drain(context.Background(), ch))
87 require.Zero(t, q.Len())
88 require.Equal(t, uint64(2), (<-ch).Flip.Seq)
89
90 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 3}})
91 q.Push(ports.OutputPresented{Flip: &ports.FlipInfo{Seq: 4}})
92 ctx, cancel := context.WithCancel(context.Background())
93 cancel()
94 ch <- ports.OutputPresented{}
95 require.ErrorIs(t, q.Drain(ctx, ch), context.Canceled)
96 q.Reset()
97 require.Zero(t, q.Len())
98}