1package broker
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10 "github.com/bnema/vev/internal/protocol"
11)
12
13// Preview push cadence and recovery. The broker owns the frame rate: local
14// previews coalesce at display speed, remote ones at a bandwidth-friendly rate.
15const (
16 previewLocalInterval = 33 * time.Millisecond
17 previewRemoteInterval = 125 * time.Millisecond
18 previewFirstFrameLimit = 3 * time.Second
19 previewReopenMin = 250 * time.Millisecond
20 previewReopenMax = 5 * time.Second
21)
22
23// SubscribePreview replaces this connection's selected-row observation after
24// validating the complete authority and its strictly increasing generation.
25func (s *Service) SubscribePreview(request ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
26 if err := request.Validate(); err != nil {
27 return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
28 }
29 if request.Epoch != s.epoch || request.Connection != s.id {
30 return nil, ports.BrokerAdmissionStale
31 }
32 s.mu.Lock()
33 if s.closed {
34 s.mu.Unlock()
35 return nil, ports.BrokerAdmissionClosed
36 }
37 if request.Generation <= s.previewGeneration {
38 s.mu.Unlock()
39 return nil, ports.BrokerAdmissionStale
40 }
41 previous := s.preview
42 s.previewGeneration = request.Generation
43 s.preview = nil
44 s.mu.Unlock()
45 if previous != nil {
46 previous.Close()
47 }
48
49 ctx, cancel := context.WithCancel(s.ctx)
50 sub := &servicePreviewSubscription{
51 service: s, request: request, ctx: ctx, cancel: cancel,
52 changed: make(chan struct{}, 1),
53 latest: previewPublication(request, protocol.RemotePreview{}, previewUnavailable("preview pending", nil)),
54 }
55 s.mu.Lock()
56 if s.closed || s.previewGeneration != request.Generation {
57 s.mu.Unlock()
58 sub.Close()
59 return nil, ports.BrokerAdmissionStale
60 }
61 s.preview = sub
62 s.wg.Add(1)
63 s.mu.Unlock()
64 go func() { defer s.wg.Done(); sub.run() }()
65 return sub, nil
66}
67
68// previewIdentifiesTarget rejects a daemon preview whose inner identity does
69// not name the requested session lifecycle and live tab. A non-OK status
70// carries no inner identity and is always accepted: it is already typed as
71// unavailable rather than as another tab's content. Successful previews
72// report viewport-maximum dimensions from the daemon crop, so inner
73// dimensions must fit inside the request instead of matching it exactly.
74func previewIdentifiesTarget(preview protocol.RemotePreview, request protocol.RemotePreviewRequest) bool {
75 if preview.Status != protocol.RemotePreviewOK {
76 return true
77 }
78 return preview.LifecycleID == request.Target.LifecycleID &&
79 preview.TabID == request.Target.LiveTabID &&
80 preview.Width != 0 && preview.Height != 0 &&
81 preview.Width <= request.Width && preview.Height <= request.Height
82}
83
84func previewUnavailable(text string, cause error) error {
85 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: text, Cause: cause}
86}
87
88func previewPublication(request ports.BrokerPreviewRequest, preview protocol.RemotePreview, err error) ports.BrokerPreviewPublication {
89 return ports.BrokerPreviewPublication{Epoch: request.Epoch, Connection: request.Connection, Generation: request.Generation,
90 Target: request.Preview.Target, Width: request.Preview.Width, Height: request.Preview.Height, Preview: preview, Err: err}
91}
92
93// previewStreamRequest builds the observation stream for one watch. The broker
94// allocates the stream identity, so a client never names one.
95func (s *Service) previewStreamRequest(route ports.BrokerPreviewRoute) (ports.BrokerOpenStreamRequest, error) {
96 s.mu.Lock()
97 defer s.mu.Unlock()
98 if s.closed {
99 return ports.BrokerOpenStreamRequest{}, ports.BrokerAdmissionClosed
100 }
101 if s.previewStream == ^ports.BrokerStreamID(0) {
102 return ports.BrokerOpenStreamRequest{}, ports.BrokerAdmissionLimit
103 }
104 s.previewStream++
105 return ports.BrokerOpenStreamRequest{
106 Epoch: s.epoch, Connection: s.previewID, Stream: s.previewStream,
107 Purpose: ports.BrokerStreamObservation, StartMode: ports.BrokerDaemonExistingOnly,
108 Local: route.Local, Endpoint: route.Endpoint, Registration: route.Registration, Policy: route.Policy,
109 }, nil
110}
111
112type servicePreviewSubscription struct {
113 service *Service
114 request ports.BrokerPreviewRequest
115 ctx context.Context
116 cancel context.CancelFunc
117 mu sync.Mutex
118 latest ports.BrokerPreviewPublication
119 changed chan struct{}
120 once sync.Once
121}
122
123func (p *servicePreviewSubscription) Changed() <-chan struct{} { return p.changed }
124func (p *servicePreviewSubscription) Latest() ports.BrokerPreviewPublication {
125 p.mu.Lock()
126 defer p.mu.Unlock()
127 return p.latest
128}
129
130// Close never waits: cancellation closes the live stream, which unblocks the
131// watch goroutine. The service wait group joins it.
132func (p *servicePreviewSubscription) Close() {
133 p.once.Do(func() {
134 p.cancel()
135 p.service.mu.Lock()
136 if p.service.preview == p {
137 p.service.preview = nil
138 }
139 p.service.mu.Unlock()
140 })
141}
142
143func (p *servicePreviewSubscription) current() bool {
144 if p.ctx.Err() != nil {
145 return false
146 }
147 p.service.mu.Lock()
148 defer p.service.mu.Unlock()
149 return !p.service.closed && p.service.preview == p && p.service.previewGeneration == p.request.Generation
150}
151
152func (p *servicePreviewSubscription) publish(preview protocol.RemotePreview, err error) bool {
153 if !p.current() {
154 return false
155 }
156 publication := previewPublication(p.request, preview, err)
157 if publication.Validate() != nil {
158 return false
159 }
160 p.mu.Lock()
161 p.latest = publication
162 p.mu.Unlock()
163 select {
164 case p.changed <- struct{}{}:
165 default:
166 }
167 return true
168}
169
170// run keeps one daemon watch stream open for the subscription's lifetime and
171// reopens it with a bounded backoff after a failure. A dead target ends the
172// subscription's recovery: there is nothing left to watch.
173func (p *servicePreviewSubscription) run() {
174 backoff := previewReopenMin
175 for p.current() {
176 reopen, delivered := p.watch()
177 if !reopen || !p.current() {
178 return
179 }
180 if delivered {
181 backoff = previewReopenMin
182 }
183 timer := p.service.clock.NewTimer(backoff)
184 select {
185 case <-p.ctx.Done():
186 timer.Stop()
187 return
188 case <-timer.C():
189 }
190 backoff = min(2*backoff, previewReopenMax)
191 }
192}
193
194// watch runs one stream until it fails. It reports whether the stream should
195// be reopened and whether it delivered at least one frame.
196func (p *servicePreviewSubscription) watch() (reopen, delivered bool) {
197 interval := previewRemoteInterval
198 if p.request.Route.Local {
199 interval = previewLocalInterval
200 }
201 route, err := p.service.previewStreamRequest(p.request.Route)
202 var stream ports.BrokerEnvelopeStream
203 if err == nil {
204 stream, err = p.service.pool.OpenStream(p.ctx, route)
205 }
206 if err != nil {
207 return p.fail(err), false
208 }
209 defer func() { _ = stream.Close() }()
210 stop := context.AfterFunc(p.ctx, func() { _ = stream.Close() })
211 defer stop()
212 typed := p.service.codec.Client(stream)
213 if err := typed.SendClient(protocol.RemotePreviewWatch{Request: p.request.Preview, MinInterval: interval}); err != nil {
214 return p.fail(err), false
215 }
216
217 // The first-frame deadline only reports the stall; the stream keeps
218 // waiting, so a slow host still delivers once it answers.
219 firstFrame := make(chan struct{})
220 deadlineDone := make(chan struct{})
221 timer := p.service.clock.NewTimer(previewFirstFrameLimit)
222 go func() {
223 defer close(deadlineDone)
224 select {
225 case <-timer.C():
226 p.publish(protocol.RemotePreview{}, previewUnavailable("preview timed out", ports.BrokerError{Code: ports.BrokerErrorTimeout}))
227 case <-firstFrame:
228 timer.Stop()
229 }
230 }()
231 var firstOnce sync.Once
232 markFirst := func() { firstOnce.Do(func() { close(firstFrame) }) }
233 defer func() { markFirst(); <-deadlineDone }()
234
235 for {
236 message, err := typed.ReceiveServer()
237 if err != nil {
238 return p.fail(err), delivered
239 }
240 preview, ok := message.(protocol.RemotePreview)
241 if !ok || protocol.ValidateRemotePreview(preview) != nil {
242 return p.fail(errors.New("broker: invalid preview response")), delivered
243 }
244 if !previewIdentifiesTarget(preview, p.request.Preview) {
245 return p.fail(errors.New("broker: preview response is for another target")), delivered
246 }
247 markFirst()
248 delivered = true
249 if !p.publish(preview, nil) {
250 return false, delivered
251 }
252 if preview.Status == protocol.RemotePreviewNoSuchTarget {
253 return false, delivered
254 }
255 }
256}
257
258// fail publishes one typed unavailable failure and reports whether recovery
259// should continue. Cancellation is the owner leaving, not a failure.
260func (p *servicePreviewSubscription) fail(err error) bool {
261 if !p.current() || errors.Is(err, context.Canceled) {
262 return false
263 }
264 return p.publish(protocol.RemotePreview{}, previewUnavailable("preview unavailable", err))
265}