serialized.go RawHistoryFolder

1package observability
2
3import (
4	"sync"
5
6	"github.com/bnema/vev/internal/ports"
7)
8
9// NewSerialized isolates producers from a potentially slow observer with one
10// ordered worker and a bounded queue.
11func NewSerialized(observer ports.RuntimeObserver, capacity int) ports.SerializedRuntimeObserver {
12	if observer == nil {
13		return nil
14	}
15	if capacity < 1 {
16		capacity = 1
17	}
18	o := &serializedRuntimeObserver{observer: observer, capacity: capacity, done: make(chan struct{})}
19	o.cond = sync.NewCond(&o.mu)
20	go o.run()
21	return o
22}
23
24// EnsureSerialized reuses an existing serialized owner or creates one.
25func EnsureSerialized(observer ports.RuntimeObserver, capacity int) (serialized ports.SerializedRuntimeObserver, owned bool) {
26	if observer == nil {
27		return nil, false
28	}
29	if serialized, ok := observer.(ports.SerializedRuntimeObserver); ok {
30		return serialized, false
31	}
32	return NewSerialized(observer, capacity), true
33}
34
35type serializedRuntimeObserver struct {
36	observer ports.RuntimeObserver
37	capacity int
38
39	mu      sync.Mutex
40	cond    *sync.Cond
41	marks   []ports.RuntimeMark
42	dropped uint64
43	closed  bool
44	active  bool
45	done    chan struct{}
46}
47
48func (o *serializedRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
49	if o == nil {
50		return
51	}
52	o.mu.Lock()
53	defer o.mu.Unlock()
54	if o.closed {
55		return
56	}
57	if o.dropped != 0 {
58		if len(o.marks)+2 > o.capacity {
59			o.dropped++
60			return
61		}
62		o.marks = append(o.marks, gapMark(o.dropped), mark)
63		o.dropped = 0
64		o.cond.Signal()
65		return
66	}
67	if len(o.marks) == o.capacity {
68		o.dropped = 1
69		return
70	}
71	o.marks = append(o.marks, mark)
72	o.cond.Signal()
73}
74
75func gapMark(dropped uint64) ports.RuntimeMark {
76	return ports.NewRuntimeMark("observer", ports.RuntimeTransportDiagnostic, dropped, false)
77}
78
79func (o *serializedRuntimeObserver) Flush() {
80	if o == nil {
81		return
82	}
83	o.mu.Lock()
84	for len(o.marks) != 0 || o.active || o.dropped != 0 {
85		o.cond.Signal()
86		o.cond.Wait()
87	}
88	closed := o.closed
89	o.mu.Unlock()
90	if closed {
91		<-o.done
92	}
93}
94
95func (o *serializedRuntimeObserver) Close() {
96	if o == nil {
97		return
98	}
99	o.mu.Lock()
100	if !o.closed {
101		o.closed = true
102		o.cond.Broadcast()
103	}
104	o.mu.Unlock()
105	<-o.done
106}
107
108func (o *serializedRuntimeObserver) run() {
109	defer close(o.done)
110	for {
111		o.mu.Lock()
112		for len(o.marks) == 0 && o.dropped == 0 && !o.closed {
113			o.cond.Wait()
114		}
115		if len(o.marks) == 0 && o.dropped != 0 {
116			o.marks = append(o.marks, gapMark(o.dropped))
117			o.dropped = 0
118		}
119		if len(o.marks) == 0 && o.closed {
120			o.cond.Broadcast()
121			o.mu.Unlock()
122			return
123		}
124		mark := o.marks[0]
125		o.marks = o.marks[1:]
126		o.active = true
127		o.mu.Unlock()
128
129		o.observer.ObserveRuntime(mark)
130
131		o.mu.Lock()
132		o.active = false
133		o.cond.Broadcast()
134		o.mu.Unlock()
135	}
136}