1package observability
2
3import (
4 "sync"
5 "testing"
6 "time"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11type runtimeObserverFunc func(ports.RuntimeMark)
12
13func (f runtimeObserverFunc) ObserveRuntime(mark ports.RuntimeMark) { f(mark) }
14
15func TestSerializedFlushesInOrderAndCloses(t *testing.T) {
16 var mu sync.Mutex
17 var got []ports.RuntimeMark
18 observer := NewSerialized(runtimeObserverFunc(func(mark ports.RuntimeMark) {
19 mu.Lock()
20 got = append(got, mark)
21 mu.Unlock()
22 }), 2)
23 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeEmitStart, 1, true))
24 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeEmitEnd, 1, true))
25 observer.Flush()
26 observer.Close()
27 observer.Close()
28
29 mu.Lock()
30 defer mu.Unlock()
31 if len(got) != 2 || got[0].Kind != ports.RuntimeEmitStart || got[1].Kind != ports.RuntimeEmitEnd {
32 t.Fatalf("serialized marks = %#v", got)
33 }
34}
35
36func TestEnsureSerializedReusesExistingOwner(t *testing.T) {
37 existing := NewSerialized(runtimeObserverFunc(func(ports.RuntimeMark) {}), 1)
38 reporter, owned := EnsureSerialized(existing, 1)
39 if owned || reporter != existing {
40 t.Fatalf("EnsureSerialized() = (%T, %t), want existing unowned reporter", reporter, owned)
41 }
42 existing.Close()
43}
44
45func TestSerializedReportsBoundedQueueLoss(t *testing.T) {
46 entered := make(chan struct{})
47 release := make(chan struct{})
48 var mu sync.Mutex
49 var got []ports.RuntimeMark
50 observer := NewSerialized(runtimeObserverFunc(func(mark ports.RuntimeMark) {
51 if mark.Kind == ports.RuntimeEmitStart {
52 close(entered)
53 <-release
54 }
55 mu.Lock()
56 got = append(got, mark)
57 mu.Unlock()
58 }), 2)
59 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeEmitStart, 1, true))
60 select {
61 case <-entered:
62 case <-time.After(time.Second):
63 t.Fatal("observer did not start")
64 }
65 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeComposeStart, 1, true))
66 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeComposeEnd, 1, true))
67 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeEmitEnd, 1, true))
68 close(release)
69 observer.Flush()
70 observer.Close()
71
72 mu.Lock()
73 defer mu.Unlock()
74 if len(got) != 4 || got[0].Kind != ports.RuntimeEmitStart || got[1].Kind != ports.RuntimeComposeStart || got[2].Kind != ports.RuntimeComposeEnd || got[3].Kind != ports.RuntimeTransportDiagnostic || got[3].Bytes != 1 || got[3].Valid {
75 t.Fatalf("bounded-loss marks = %#v", got)
76 }
77}