1package daemon
2
3import (
4 "errors"
5 "sync"
6 "sync/atomic"
7
8 renderer "github.com/bnema/vev-vt/ansi"
9 "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/internal/ports"
11)
12
13// outputStateStream owns the terminal-output dependency chain for one
14// attached client. Each incremental output is rendered from the preceding
15// emitted frame and therefore remains pipelineable without waiting for an
16// ACK. Cumulative ACKs advance the bounded paint window.
17// A reset frame starts a new epoch, clears the dependency chain, and advertises
18// state 0 as its base.
19//
20// Callers serialize access with attachedClient.sendMu. A prepared transaction
21// must be sent or committed as a no-op before the next transaction is prepared.
22type outputStateStream struct {
23 renderer *renderer.Renderer
24 epoch uint64
25 next uint64
26 acked uint64
27 generation uint64
28 maxOutstanding uint64
29 outstandingAtomic atomic.Uint64
30 maxOutstandingAtomic atomic.Uint64
31 forceSnapshot bool
32 initialized bool
33 attachment *attachedClient
34 stateMu sync.Mutex
35}
36
37// lockView serializes an attachment's view publication with every output
38// state transition. It deliberately does not take sendMu: callers that hold
39// architecture locks can publish a view without reversing the paint path's
40// sendMu -> session lock order.
41func (s *outputStateStream) lockView() {
42 if s == nil {
43 return
44 }
45 if s.attachment == nil {
46 s.stateMu.Lock()
47 return
48 }
49 s.attachment.viewMu.Lock()
50}
51
52func (s *outputStateStream) unlockView() {
53 if s == nil {
54 return
55 }
56 if s.attachment == nil {
57 s.stateMu.Unlock()
58 return
59 }
60 s.attachment.viewMu.Unlock()
61}
62
63func newOutputStateStream(windowSize ...uint8) *outputStateStream {
64 window := uint8(maxUnackedOutputStates)
65 if len(windowSize) > 0 {
66 window = normalizeOutputWindow(windowSize[0])
67 }
68 stream := &outputStateStream{
69 renderer: renderer.New(renderer.Capabilities{}),
70 epoch: 1,
71 maxOutstanding: uint64(window),
72 }
73 stream.maxOutstandingAtomic.Store(uint64(window))
74 return stream
75}
76
77type preparedOutput struct {
78 stream *outputStateStream
79 draw renderer.PreparedDraw
80 base uint64
81 next uint64
82 epoch uint64
83 generation uint64
84 connectionGeneration uint64
85 viewRevision uint64
86 size domain.Size
87 data []byte
88 reset bool
89 attempted bool
90 sent bool
91}
92
93func (s *outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (*preparedOutput, error) {
94 s.lockView()
95 defer s.unlockView()
96 if s.epoch == 0 {
97 s.epoch = 1
98 }
99 if reset && s.initialized {
100 s.rebaseLocked()
101 }
102 reset = reset || s.forceSnapshot || !s.initialized
103 if s.next == ^uint64(0) {
104 return nil, errors.New("output state number exhausted")
105 }
106 draw, err := s.renderer.Prepare(frame, damage, reset)
107 if err != nil {
108 return nil, err
109 }
110 s.initialized = true
111 generation, epoch, viewRevision := s.fenceLocked()
112 return &preparedOutput{
113 stream: s,
114 draw: draw,
115 base: s.next,
116 next: s.next + 1,
117 epoch: epoch,
118 generation: s.generation,
119 connectionGeneration: generation,
120 viewRevision: viewRevision,
121 size: domain.Size{Cols: frame.Width, Rows: frame.Height},
122 data: draw.Bytes(),
123 reset: reset,
124 }, nil
125}
126
127func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame) error) error {
128 if p == nil || p.stream == nil || p.attempted {
129 return nil
130 }
131 p.stream.lockView()
132 defer p.stream.unlockView()
133 p.attempted = true
134 if !p.currentLocked() {
135 return nil
136 }
137 if len(data) == 0 {
138 p.commit()
139 p.sent = true
140 return nil
141 }
142 base := p.base
143 if p.reset {
144 base = 0
145 }
146 out, err := marshalOutputState(data, p.epoch, base, p.next, echoAck, p.viewRevision, p.size, p.reset)
147 if err != nil {
148 p.stream.forceSnapshot = true
149 return err
150 }
151 if err := send(out); err != nil {
152 p.stream.forceSnapshot = true
153 return err
154 }
155 p.commit()
156 p.stream.next = p.next
157 p.stream.publishOutstanding()
158 p.sent = true
159 return nil
160}
161
162func (p *preparedOutput) commitNoSend() {
163 if p == nil || p.stream == nil || p.attempted {
164 return
165 }
166 p.stream.lockView()
167 defer p.stream.unlockView()
168 p.attempted = true
169 if !p.currentLocked() || len(p.data) != 0 {
170 return
171 }
172 p.commit()
173 p.sent = true
174}
175
176func (p *preparedOutput) commit() {
177 p.draw.Commit()
178 p.stream.forceSnapshot = false
179}
180
181func (s *outputStateStream) currentEpoch() uint64 {
182 s.lockView()
183 defer s.unlockView()
184 return s.currentEpochLocked()
185}
186
187func (s *outputStateStream) currentEpochLocked() uint64 {
188 if s.epoch == 0 {
189 s.epoch = 1
190 }
191 return s.epoch
192}
193
194func (s *outputStateStream) fenceLocked() (uint64, uint64, uint64) {
195 if s == nil {
196 return 0, 1, 0
197 }
198 generation, revision := uint64(0), uint64(0)
199 if s.attachment != nil {
200 generation = s.attachment.connectionGeneration.Load()
201 revision = s.attachment.view.revision
202 }
203 return generation, s.currentEpochLocked(), revision
204}
205
206func (s *outputStateStream) preparedCurrentLocked(generation, epoch, viewRevision, base uint64) bool {
207 if s == nil || s.currentEpochLocked() != epoch || s.generation != generation || s.next != base {
208 return false
209 }
210 if s.attachment == nil {
211 return true
212 }
213 return s.attachment.view.revision == viewRevision
214}
215
216func (p *preparedOutput) currentLocked() bool {
217 return p != nil && p.stream != nil && p.stream.preparedCurrentLocked(p.generation, p.epoch, p.viewRevision, p.base) &&
218 (p.stream.attachment == nil || p.stream.attachment.connectionGeneration.Load() == p.connectionGeneration)
219}
220
221func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision uint64, size domain.Size, full bool) (ports.Frame, error) {
222 payload, err := ports.MarshalOutput(ports.Output{
223 Epoch: epoch, Base: base, New: next, Echo: echoAck, ViewRevision: viewRevision,
224 Size: size, Full: full, Data: data,
225 })
226 if err != nil {
227 return ports.Frame{}, err
228 }
229 return ports.Frame{Type: ports.MsgOutput, Payload: payload}, nil
230}
231
232// sideEffect builds an output frame without advancing the state stream. When
233// an attachment is present, callers must hold its sendMu; attachment-bound
234// sends use sideEffectLocked after validating their transport incarnation.
235func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
236 s.lockView()
237 defer s.unlockView()
238 return s.sideEffectLocked(data, echoAck)
239}
240
241// sideEffectLocked requires the attachment view lock for both frame creation
242// and transport emission. Keeping that lock across the send prevents a view
243// publication from overtaking a frame that was fenced just before it.
244func (s *outputStateStream) sideEffectLocked(data []byte, echoAck uint64) (ports.Frame, error) {
245 _, epoch, viewRevision := s.fenceLocked()
246 size := domain.Size{Cols: 1, Rows: 1}
247 if s != nil && s.attachment != nil {
248 size = s.attachment.sizeSnapshot()
249 }
250 return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false)
251}
252
253func (s *outputStateStream) ack(epoch, state uint64) bool {
254 if s == nil {
255 return false
256 }
257 s.lockView()
258 defer s.unlockView()
259 if epoch != s.currentEpochLocked() || state > s.next || state <= s.acked {
260 return false
261 }
262 s.acked = state
263 s.publishOutstanding()
264 return true
265}
266
267func (s *outputStateStream) rebase() {
268 if s == nil {
269 return
270 }
271 s.lockView()
272 defer s.unlockView()
273 s.rebaseLocked()
274}
275
276// rebaseLocked is called while the attachment view lock is held. Publishing
277// the next view revision takes this path before making that revision visible.
278func (s *outputStateStream) rebaseLocked() {
279 if s == nil {
280 return
281 }
282 epoch := s.currentEpochLocked()
283 if epoch == ^uint64(0) {
284 return
285 }
286 s.epoch = epoch + 1
287 s.next = 0
288 s.acked = 0
289 s.generation++
290 s.publishOutstanding()
291 s.maxOutstandingAtomic.Store(s.maxOutstanding)
292 s.renderer.Reset()
293 s.forceSnapshot = true
294 s.initialized = false
295}
296
297func (s *outputStateStream) publishOutstanding() {
298 if s == nil {
299 return
300 }
301 s.outstandingAtomic.Store(s.next - s.acked)
302}
303
304// syncCapacityLocked refreshes the lock-free readiness snapshot after a
305// caller mutates the legacy fields under attachedClient.sendMu.
306func (s *outputStateStream) syncCapacityLocked() {
307 if s == nil {
308 return
309 }
310 s.lockView()
311 defer s.unlockView()
312 s.maxOutstandingAtomic.Store(s.maxOutstanding)
313 s.publishOutstanding()
314}
315
316func (s *outputStateStream) atCapacity() bool {
317 if s == nil {
318 return false
319 }
320 maxOutstanding := s.maxOutstandingAtomic.Load()
321 if maxOutstanding == 0 {
322 maxOutstanding = s.maxOutstanding
323 }
324 return s.outstandingAtomic.Load() >= maxOutstanding
325}
326
327func (s *outputStateStream) outstanding() uint64 {
328 if s == nil {
329 return 0
330 }
331 s.lockView()
332 defer s.unlockView()
333 return s.next - s.acked
334}