1package client
2
3import (
4 "context"
5 "errors"
6 "sync"
7
8 "github.com/bnema/vev/internal/ports"
9 "github.com/bnema/vev/internal/protocol"
10)
11
12// Outbox bounds. Past either, enqueue blocks until the sender makes progress
13// or the caller is cancelled. Input is acknowledged to the terminal pump only
14// once written, so the pump stops reading new keys long before these limits;
15// they bound acks, focus reports, and other control traffic.
16const (
17 outboxMaxPending = 4096
18 outboxMaxBytes = 4 << 20
19)
20
21// errOutboxStopped is returned by SendClient after the outbox stopped.
22var errOutboxStopped = errors.New("client: attachment outbox stopped")
23
24// attachmentOutbox decouples the attachment worker loop from the carriage:
25// enqueue returns once the message is accepted, and one sender goroutine
26// writes messages in order. A stalled link therefore no longer freezes key
27// handling, overlays, or painting. The first write failure is sticky: it is
28// reported on Failed and returned by every later enqueue.
29type attachmentOutbox struct {
30 ports.BrokerLogicalConnection
31
32 mu sync.Mutex
33 queue []protocol.ClientMessage
34 bytes int // payload bytes queued
35 queued uint64 // messages ever accepted
36 written uint64 // messages written successfully
37 err error
38 stopped bool
39 wake chan struct{} // sender wake-up, capacity 1
40 space chan struct{} // closed-and-replaced after each write, closed on failure
41 idle chan struct{} // closed-and-replaced when the queue drains
42 failed chan struct{} // closed once on the first write failure
43 done chan struct{} // closed when the sender exits
44}
45
46func newAttachmentOutbox(stream ports.BrokerLogicalConnection) *attachmentOutbox {
47 b := &attachmentOutbox{
48 BrokerLogicalConnection: stream,
49 wake: make(chan struct{}, 1),
50 space: make(chan struct{}),
51 idle: make(chan struct{}),
52 failed: make(chan struct{}),
53 done: make(chan struct{}),
54 }
55 close(b.idle)
56 go b.run()
57 return b
58}
59
60// outboxPayload is the variable-size part of a message counted against
61// outboxMaxBytes.
62func outboxPayload(message protocol.ClientMessage) int {
63 switch m := message.(type) {
64 case protocol.Input:
65 return len(m.Data)
66 case protocol.ImagePush:
67 return len(m.Data)
68 }
69 return 0
70}
71
72// SendClient enqueues message without a cancellation source.
73func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
74 return b.enqueue(context.Background(), nil, message)
75}
76
77// enqueue accepts message for ordered transmission. It blocks only while the
78// outbox is full, until space frees, ctx ends, or cancel closes. It returns
79// the sticky failure once the sender failed.
80func (b *attachmentOutbox) enqueue(ctx context.Context, cancel <-chan struct{}, message protocol.ClientMessage) error {
81 size := outboxPayload(message)
82 b.mu.Lock()
83 for {
84 if b.err != nil {
85 err := b.err
86 b.mu.Unlock()
87 return err
88 }
89 if b.stopped {
90 b.mu.Unlock()
91 return errOutboxStopped
92 }
93 // An empty queue always admits, so one oversized message still goes.
94 if len(b.queue) == 0 || (len(b.queue) < outboxMaxPending && b.bytes+size <= outboxMaxBytes) {
95 break
96 }
97 space := b.space
98 b.mu.Unlock()
99 select {
100 case <-space:
101 case <-b.done:
102 case <-ctx.Done():
103 return ctx.Err()
104 case <-cancel:
105 return errAttachmentForegroundRevoked
106 }
107 b.mu.Lock()
108 }
109 if len(b.queue) == 0 {
110 b.idle = make(chan struct{})
111 }
112 b.queue = append(b.queue, message)
113 b.bytes += size
114 b.queued++
115 b.mu.Unlock()
116 select {
117 case b.wake <- struct{}{}:
118 default:
119 }
120 return nil
121}
122
123// Close stops the sender and closes the underlying stream, which unblocks a
124// write stalled on the link.
125func (b *attachmentOutbox) Close() error {
126 b.stop()
127 return b.BrokerLogicalConnection.Close()
128}
129
130// stop ends the sender after its current write; queued messages are dropped.
131// Session input among them was never acknowledged to the terminal pump, so
132// the worker preserves it like a failed synchronous send.
133func (b *attachmentOutbox) stop() {
134 b.mu.Lock()
135 b.stopped = true
136 b.mu.Unlock()
137 select {
138 case b.wake <- struct{}{}:
139 default:
140 }
141}
142
143// unsentInput returns the session input bytes of accepted messages numbered
144// (from, to] that were never written. It is final only after join.
145func (b *attachmentOutbox) unsentInput(from, to uint64) []byte {
146 b.mu.Lock()
147 defer b.mu.Unlock()
148 var data []byte
149 // queue[i] is accepted message number written+1+i.
150 for i, message := range b.queue {
151 n := b.written + 1 + uint64(i)
152 if n <= from || n > to {
153 continue
154 }
155 if input, ok := message.(protocol.Input); ok {
156 data = append(data, input.Data...)
157 }
158 }
159 return data
160}
161
162// join stops the sender, closes the stream so a stalled write returns, and
163// waits for the sender to exit. Afterwards writtenCount is final.
164func (b *attachmentOutbox) join() {
165 _ = b.Close()
166 <-b.done
167}
168
169// accepted reports how many messages were ever accepted.
170func (b *attachmentOutbox) accepted() uint64 {
171 b.mu.Lock()
172 defer b.mu.Unlock()
173 return b.queued
174}
175
176// writtenCount reports how many messages were written successfully.
177func (b *attachmentOutbox) writtenCount() uint64 {
178 b.mu.Lock()
179 defer b.mu.Unlock()
180 return b.written
181}
182
183// Progress is closed after the next successful write or on failure.
184func (b *attachmentOutbox) Progress() <-chan struct{} {
185 b.mu.Lock()
186 defer b.mu.Unlock()
187 return b.space
188}
189
190// Failed is closed once a write failed; failure returns the cause.
191func (b *attachmentOutbox) Failed() <-chan struct{} { return b.failed }
192
193func (b *attachmentOutbox) failure() error {
194 b.mu.Lock()
195 defer b.mu.Unlock()
196 return b.err
197}
198
199// Idle is closed while nothing is queued or being written.
200func (b *attachmentOutbox) Idle() <-chan struct{} {
201 b.mu.Lock()
202 defer b.mu.Unlock()
203 return b.idle
204}
205
206func (b *attachmentOutbox) run() {
207 defer close(b.done)
208 for {
209 b.mu.Lock()
210 if b.stopped {
211 b.mu.Unlock()
212 return
213 }
214 if len(b.queue) == 0 {
215 b.mu.Unlock()
216 <-b.wake
217 continue
218 }
219 message := b.queue[0]
220 b.mu.Unlock()
221
222 err := b.BrokerLogicalConnection.SendClient(message)
223
224 b.mu.Lock()
225 if err != nil {
226 // The queue is kept, failed write first, so unsentInput can
227 // hand undelivered keys back to the terminal pump.
228 b.err = err
229 close(b.failed)
230 close(b.idle)
231 close(b.space)
232 b.mu.Unlock()
233 return
234 }
235 b.queue[0] = nil
236 b.queue = b.queue[1:]
237 b.bytes -= outboxPayload(message)
238 b.written++
239 close(b.space)
240 b.space = make(chan struct{})
241 if len(b.queue) == 0 {
242 close(b.idle)
243 }
244 b.mu.Unlock()
245 }
246}