fix(core): schedule recording by frame slots and preserve real failures
- commit
- 8984092c81a26fc6da57d3ef5a9f3ac5aa4fb831
- author
- brice · · 2026-09-30 08:14 UTC
- committer
- brice · · 2026-09-30 14:48 UTC
- parent
- e272971
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +316 -73
M internal/adapters/cli/cli.go +4 -4
| @@ -43,7 +43,7 @@ type Options struct { | |||
| 43 | 43 | ||
| 44 | 44 | // Usage describes the command line. | |
| 45 | 45 | func Usage() string { | |
| 46 | - | return `Usage: nefercap [command] [flags] | |
| 46 | + | return fmt.Sprintf(`Usage: nefercap [command] [flags] | |
| 47 | 47 | ||
| 48 | 48 | Commands: | |
| 49 | 49 | gui open the capture control panel (default) | |
| @@ -56,10 +56,10 @@ Flags: | |||
| 56 | 56 | -output NAME output name (screenshot, record) | |
| 57 | 57 | -region X,Y,WxH output-local logical region; default is the full output (screenshot, record) | |
| 58 | 58 | -file PATH new file to write, required (screenshot, record) | |
| 59 | - | -fps N frames per second, 1..120 (record, default 30) | |
| 60 | - | -size WxH scaled video size, both even (record, default source size) | |
| 59 | + | -fps N frames per second, 1..%d (record, default %d) | |
| 60 | + | -size WxH scaled video size, both even, at most %d each (record, default source size) | |
| 61 | 61 | -duration D recording length such as 30s; 0 records until interrupted (record) | |
| 62 | - | ` | |
| 62 | + | `, ports.MaxFPS, DefaultFPS, ports.MaxDimension) | |
| 63 | 63 | } | |
| 64 | 64 | ||
| 65 | 65 | // Parse parses args (without the program name). Help requests print Usage to | |
| @@ -43,7 +43,7 @@ type Options struct { | |||||
| 43 | 43 | ||||
| 44 | // Usage describes the command line. | 44 | // Usage describes the command line. | ||
| 45 | func Usage() string { | 45 | func Usage() string { | ||
| 46 | - | return `Usage: nefercap [command] [flags] | 46 | + | return fmt.Sprintf(`Usage: nefercap [command] [flags] |
| 47 | 47 | ||||
| 48 | Commands: | 48 | Commands: | ||
| 49 | gui open the capture control panel (default) | 49 | gui open the capture control panel (default) | ||
| @@ -56,10 +56,10 @@ Flags: | |||||
| 56 | -output NAME output name (screenshot, record) | 56 | -output NAME output name (screenshot, record) | ||
| 57 | -region X,Y,WxH output-local logical region; default is the full output (screenshot, record) | 57 | -region X,Y,WxH output-local logical region; default is the full output (screenshot, record) | ||
| 58 | -file PATH new file to write, required (screenshot, record) | 58 | -file PATH new file to write, required (screenshot, record) | ||
| 59 | - | -fps N frames per second, 1..120 (record, default 30) | 59 | + | -fps N frames per second, 1..%d (record, default %d) |
| 60 | - | -size WxH scaled video size, both even (record, default source size) | 60 | + | -size WxH scaled video size, both even, at most %d each (record, default source size) |
| 61 | -duration D recording length such as 30s; 0 records until interrupted (record) | 61 | -duration D recording length such as 30s; 0 records until interrupted (record) | ||
| 62 | - | ` | 62 | + | `, ports.MaxFPS, DefaultFPS, ports.MaxDimension) |
| 63 | } | 63 | } | ||
| 64 | 64 | ||||
| 65 | // Parse parses args (without the program name). Help requests print Usage to | 65 | // Parse parses args (without the program name). Help requests print Usage to | ||
M internal/core/service.go +138 -34
| @@ -59,56 +59,121 @@ func (s *Service) screenshot(ctx context.Context, sel ports.Selection) error { | |||
| 59 | 59 | } | |
| 60 | 60 | ||
| 61 | 61 | func (s *Service) record(ctx context.Context, sel ports.Selection) error { | |
| 62 | - | runCtx := ctx | |
| 63 | - | if sel.Duration > 0 { | |
| 64 | - | var cancel context.CancelFunc | |
| 65 | - | runCtx, cancel = context.WithTimeout(ctx, sel.Duration) | |
| 66 | - | defer cancel() | |
| 67 | - | } | |
| 68 | - | ||
| 69 | - | frame, err := s.source.Capture(runCtx, sel.Target) | |
| 62 | + | frame, err := s.source.Capture(ctx, sel.Target) | |
| 70 | 63 | if err != nil { | |
| 71 | 64 | return fmt.Errorf("capture first frame: %w", err) | |
| 72 | 65 | } | |
| 73 | 66 | if err := frame.Validate(); err != nil { | |
| 74 | 67 | return fmt.Errorf("captured frame: %w", err) | |
| 75 | 68 | } | |
| 76 | - | if err := s.video.Start(runCtx, frame, sel.Path, sel.Video); err != nil { | |
| 77 | - | // Nothing was recorded: discard instead of finalizing. | |
| 78 | - | err = fmt.Errorf("start video: %w", err) | |
| 79 | - | if abortErr := s.video.Abort(); abortErr != nil { | |
| 80 | - | err = errors.Join(err, fmt.Errorf("abort video: %w", abortErr)) | |
| 81 | - | } | |
| 82 | - | return err | |
| 69 | + | if err := s.video.Start(ctx, frame, sel.Path, sel.Video); err != nil { | |
| 70 | + | return s.abort(fmt.Errorf("start video: %w", err)) | |
| 71 | + | } | |
| 72 | + | if err := s.video.Write(ctx, frame); err != nil { | |
| 73 | + | // Nothing was recorded, even when the cause is cancellation: discard | |
| 74 | + | // the file instead of finalizing an empty recording. | |
| 75 | + | return s.abort(fmt.Errorf("write first frame: %w", err)) | |
| 83 | 76 | } | |
| 84 | 77 | ||
| 85 | - | loopErr := s.recordLoop(runCtx, frame, sel) | |
| 86 | - | // Graceful finalize on every exit after Start; the writer bounds it. | |
| 78 | + | loopErr := s.recordLoop(ctx, frame, sel) | |
| 79 | + | // Graceful finalize on every exit after the first frame; the writer bounds it. | |
| 87 | 80 | if closeErr := s.video.Close(); closeErr != nil { | |
| 88 | 81 | return errors.Join(loopErr, fmt.Errorf("finalize video: %w", closeErr)) | |
| 89 | 82 | } | |
| 90 | 83 | return loopErr | |
| 91 | 84 | } | |
| 92 | 85 | ||
| 93 | - | // recordLoop writes the first frame, then one frame per tick. It returns nil | |
| 94 | - | // when the run context ends (duration elapsed or cancellation). | |
| 95 | - | func (s *Service) recordLoop(ctx context.Context, first ports.Frame, sel ports.Selection) error { | |
| 96 | - | if err := s.video.Write(ctx, first); err != nil { | |
| 97 | - | return writeResult(ctx, err) | |
| 86 | + | func (s *Service) abort(err error) error { | |
| 87 | + | if abortErr := s.video.Abort(); abortErr != nil { | |
| 88 | + | return errors.Join(err, fmt.Errorf("abort video: %w", abortErr)) | |
| 98 | 89 | } | |
| 99 | - | ticker := time.NewTicker(time.Second / time.Duration(sel.Video.FPS)) | |
| 100 | - | defer ticker.Stop() | |
| 101 | - | for { | |
| 102 | - | select { | |
| 103 | - | case <-ctx.Done(): | |
| 104 | - | return nil | |
| 105 | - | case <-ticker.C: | |
| 90 | + | return err | |
| 91 | + | } | |
| 92 | + | ||
| 93 | + | // stopWriteTimeout bounds the trailing duplicate writes made after a stop | |
| 94 | + | // request, which must not be interrupted by the cancellation itself. | |
| 95 | + | const stopWriteTimeout = 2 * time.Second | |
| 96 | + | ||
| 97 | + | // schedule maps elapsed time since the first frame to fixed-rate frame slots. | |
| 98 | + | // Slot n is due at n/fps. total is the slot count of a timed recording, or 0 | |
| 99 | + | // when recording until cancellation. | |
| 100 | + | type schedule struct { | |
| 101 | + | start time.Time | |
| 102 | + | fps int64 | |
| 103 | + | total int64 | |
| 104 | + | } | |
| 105 | + | ||
| 106 | + | func newSchedule(start time.Time, sel ports.Selection) schedule { | |
| 107 | + | sc := schedule{start: start, fps: int64(sel.Video.FPS)} | |
| 108 | + | if sel.Duration > 0 { | |
| 109 | + | // ceil(duration*fps) split into whole seconds and remainder so the | |
| 110 | + | // arithmetic cannot overflow for any positive time.Duration. | |
| 111 | + | secs, rem := int64(sel.Duration/time.Second), int64(sel.Duration%time.Second) | |
| 112 | + | sc.total = secs*sc.fps + (rem*sc.fps+int64(time.Second)-1)/int64(time.Second) | |
| 113 | + | } | |
| 114 | + | return sc | |
| 115 | + | } | |
| 116 | + | ||
| 117 | + | // due returns the offset at which slot n starts. | |
| 118 | + | func (sc schedule) due(n int64) time.Duration { | |
| 119 | + | q, r := n/sc.fps, n%sc.fps | |
| 120 | + | return time.Duration(q)*time.Second + time.Duration(r)*time.Second/time.Duration(sc.fps) | |
| 121 | + | } | |
| 122 | + | ||
| 123 | + | // current returns the slot containing now, limited to the last slot of a | |
| 124 | + | // timed recording. past reports that a timed recording's window has ended. | |
| 125 | + | func (sc schedule) current() (slot int64, past bool) { | |
| 126 | + | elapsed := time.Since(sc.start) | |
| 127 | + | secs, rem := int64(elapsed/time.Second), int64(elapsed%time.Second) | |
| 128 | + | slot = secs*sc.fps + rem*sc.fps/int64(time.Second) | |
| 129 | + | if sc.total > 0 && slot >= sc.total { | |
| 130 | + | return sc.total - 1, true | |
| 131 | + | } | |
| 132 | + | return slot, false | |
| 133 | + | } | |
| 134 | + | ||
| 135 | + | // recordLoop fills slots 1.. after the already written first frame. A slot | |
| 136 | + | // missed because Capture or Write ran long reuses the current frame, which is | |
| 137 | + | // still valid until the next Capture, so the encoded duration stays exact | |
| 138 | + | // even when wall time overruns. A timed recording ends after ceil(duration*fps) | |
| 139 | + | // frames; otherwise it ends when ctx is done. | |
| 140 | + | func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.Selection) error { | |
| 141 | + | sc := newSchedule(time.Now(), sel) | |
| 142 | + | timer := time.NewTimer(time.Hour) | |
| 143 | + | defer timer.Stop() | |
| 144 | + | timer.Stop() | |
| 145 | + | ||
| 146 | + | next := int64(1) | |
| 147 | + | for sc.total == 0 || next < sc.total { | |
| 148 | + | if wait := sc.due(next) - time.Since(sc.start); wait > 0 { | |
| 149 | + | timer.Reset(wait) | |
| 150 | + | select { | |
| 151 | + | case <-ctx.Done(): | |
| 152 | + | return s.fillElapsed(ctx, frame, next, sc) | |
| 153 | + | case <-timer.C: | |
| 154 | + | } | |
| 155 | + | } else if ctx.Err() != nil { | |
| 156 | + | return s.fillElapsed(ctx, frame, next, sc) | |
| 157 | + | } | |
| 158 | + | ||
| 159 | + | cur, past := sc.current() | |
| 160 | + | cur = max(cur, next) | |
| 161 | + | // Slots already missed repeat the frame before it is invalidated. | |
| 162 | + | last := cur - 1 | |
| 163 | + | if past { | |
| 164 | + | last = cur // window over: no capture for the final slot | |
| 165 | + | } | |
| 166 | + | for ; next <= last; next++ { | |
| 167 | + | if err := s.video.Write(ctx, frame); err != nil { | |
| 168 | + | return writeResult(ctx, err) | |
| 169 | + | } | |
| 106 | 170 | } | |
| 107 | - | if ctx.Err() != nil { | |
| 171 | + | if past { | |
| 108 | 172 | return nil | |
| 109 | 173 | } | |
| 110 | - | frame, err := s.source.Capture(ctx, sel.Target) | |
| 111 | - | if err != nil { | |
| 174 | + | ||
| 175 | + | var err error | |
| 176 | + | if frame, err = s.source.Capture(ctx, sel.Target); err != nil { | |
| 112 | 177 | if stopped(ctx, err) { | |
| 113 | 178 | return nil | |
| 114 | 179 | } | |
| @@ -120,7 +185,26 @@ func (s *Service) recordLoop(ctx context.Context, first ports.Frame, sel ports.S | |||
| 120 | 185 | if err := s.video.Write(ctx, frame); err != nil { | |
| 121 | 186 | return writeResult(ctx, err) | |
| 122 | 187 | } | |
| 188 | + | next++ | |
| 123 | 189 | } | |
| 190 | + | return nil | |
| 191 | + | } | |
| 192 | + | ||
| 193 | + | // fillElapsed repeats the current frame for slots that began before the stop | |
| 194 | + | // request so the file covers the time actually recorded. | |
| 195 | + | func (s *Service) fillElapsed(ctx context.Context, frame ports.Frame, next int64, sc schedule) error { | |
| 196 | + | cur, _ := sc.current() | |
| 197 | + | if next > cur { | |
| 198 | + | return nil | |
| 199 | + | } | |
| 200 | + | wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), stopWriteTimeout) | |
| 201 | + | defer cancel() | |
| 202 | + | for ; next <= cur; next++ { | |
| 203 | + | if err := s.video.Write(wctx, frame); err != nil { | |
| 204 | + | return fmt.Errorf("write trailing frame: %w", err) | |
| 205 | + | } | |
| 206 | + | } | |
| 207 | + | return nil | |
| 124 | 208 | } | |
| 125 | 209 | ||
| 126 | 210 | func writeResult(ctx context.Context, err error) error { | |
| @@ -130,9 +214,29 @@ func writeResult(ctx context.Context, err error) error { | |||
| 130 | 214 | return fmt.Errorf("write frame: %w", err) | |
| 131 | 215 | } | |
| 132 | 216 | ||
| 133 | - | // stopped reports whether err is only the consequence of the run ending. | |
| 217 | + | // stopped reports whether the run is ending and err consists only of context | |
| 218 | + | // cancellation or deadline leaves. A failure joined with cancellation is real. | |
| 134 | 219 | func stopped(ctx context.Context, err error) bool { | |
| 135 | - | return ctx.Err() != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) | |
| 220 | + | return ctx.Err() != nil && onlyContext(err) | |
| 221 | + | } | |
| 222 | + | ||
| 223 | + | func onlyContext(err error) bool { | |
| 224 | + | switch e := err.(type) { | |
| 225 | + | case nil: | |
| 226 | + | return false | |
| 227 | + | case interface{ Unwrap() []error }: | |
| 228 | + | children := e.Unwrap() | |
| 229 | + | for _, c := range children { | |
| 230 | + | if !onlyContext(c) { | |
| 231 | + | return false | |
| 232 | + | } | |
| 233 | + | } | |
| 234 | + | return len(children) > 0 | |
| 235 | + | case interface{ Unwrap() error }: | |
| 236 | + | return err == context.Canceled || err == context.DeadlineExceeded || onlyContext(e.Unwrap()) | |
| 237 | + | default: | |
| 238 | + | return err == context.Canceled || err == context.DeadlineExceeded | |
| 239 | + | } | |
| 136 | 240 | } | |
| 137 | 241 | ||
| 138 | 242 | func validateSelection(sel ports.Selection) error { | |
| @@ -59,56 +59,121 @@ func (s *Service) screenshot(ctx context.Context, sel ports.Selection) error { | |||||
| 59 | } | 59 | } | ||
| 60 | 60 | ||||
| 61 | func (s *Service) record(ctx context.Context, sel ports.Selection) error { | 61 | func (s *Service) record(ctx context.Context, sel ports.Selection) error { | ||
| 62 | - | runCtx := ctx | 62 | + | frame, err := s.source.Capture(ctx, sel.Target) |
| 63 | - | if sel.Duration > 0 { | |||
| 64 | - | var cancel context.CancelFunc | |||
| 65 | - | runCtx, cancel = context.WithTimeout(ctx, sel.Duration) | |||
| 66 | - | defer cancel() | |||
| 67 | - | } | |||
| 68 | - | ||||
| 69 | - | frame, err := s.source.Capture(runCtx, sel.Target) | |||
| 70 | if err != nil { | 63 | if err != nil { | ||
| 71 | return fmt.Errorf("capture first frame: %w", err) | 64 | return fmt.Errorf("capture first frame: %w", err) | ||
| 72 | } | 65 | } | ||
| 73 | if err := frame.Validate(); err != nil { | 66 | if err := frame.Validate(); err != nil { | ||
| 74 | return fmt.Errorf("captured frame: %w", err) | 67 | return fmt.Errorf("captured frame: %w", err) | ||
| 75 | } | 68 | } | ||
| 76 | - | if err := s.video.Start(runCtx, frame, sel.Path, sel.Video); err != nil { | 69 | + | if err := s.video.Start(ctx, frame, sel.Path, sel.Video); err != nil { |
| 77 | - | // Nothing was recorded: discard instead of finalizing. | 70 | + | return s.abort(fmt.Errorf("start video: %w", err)) |
| 78 | - | err = fmt.Errorf("start video: %w", err) | 71 | + | } |
| 79 | - | if abortErr := s.video.Abort(); abortErr != nil { | 72 | + | if err := s.video.Write(ctx, frame); err != nil { |
| 80 | - | err = errors.Join(err, fmt.Errorf("abort video: %w", abortErr)) | 73 | + | // Nothing was recorded, even when the cause is cancellation: discard |
| 81 | - | } | 74 | + | // the file instead of finalizing an empty recording. |
| 82 | - | return err | 75 | + | return s.abort(fmt.Errorf("write first frame: %w", err)) |
| 83 | } | 76 | } | ||
| 84 | 77 | ||||
| 85 | - | loopErr := s.recordLoop(runCtx, frame, sel) | 78 | + | loopErr := s.recordLoop(ctx, frame, sel) |
| 86 | - | // Graceful finalize on every exit after Start; the writer bounds it. | 79 | + | // Graceful finalize on every exit after the first frame; the writer bounds it. |
| 87 | if closeErr := s.video.Close(); closeErr != nil { | 80 | if closeErr := s.video.Close(); closeErr != nil { | ||
| 88 | return errors.Join(loopErr, fmt.Errorf("finalize video: %w", closeErr)) | 81 | return errors.Join(loopErr, fmt.Errorf("finalize video: %w", closeErr)) | ||
| 89 | } | 82 | } | ||
| 90 | return loopErr | 83 | return loopErr | ||
| 91 | } | 84 | } | ||
| 92 | 85 | ||||
| 93 | - | // recordLoop writes the first frame, then one frame per tick. It returns nil | 86 | + | func (s *Service) abort(err error) error { |
| 94 | - | // when the run context ends (duration elapsed or cancellation). | 87 | + | if abortErr := s.video.Abort(); abortErr != nil { |
| 95 | - | func (s *Service) recordLoop(ctx context.Context, first ports.Frame, sel ports.Selection) error { | 88 | + | return errors.Join(err, fmt.Errorf("abort video: %w", abortErr)) |
| 96 | - | if err := s.video.Write(ctx, first); err != nil { | |||
| 97 | - | return writeResult(ctx, err) | |||
| 98 | } | 89 | } | ||
| 99 | - | ticker := time.NewTicker(time.Second / time.Duration(sel.Video.FPS)) | 90 | + | return err |
| 100 | - | defer ticker.Stop() | 91 | + | } |
| 101 | - | for { | 92 | + | |
| 102 | - | select { | 93 | + | // stopWriteTimeout bounds the trailing duplicate writes made after a stop |
| 103 | - | case <-ctx.Done(): | 94 | + | // request, which must not be interrupted by the cancellation itself. |
| 104 | - | return nil | 95 | + | const stopWriteTimeout = 2 * time.Second |
| 105 | - | case <-ticker.C: | 96 | + | |
| 97 | + | // schedule maps elapsed time since the first frame to fixed-rate frame slots. | |||
| 98 | + | // Slot n is due at n/fps. total is the slot count of a timed recording, or 0 | |||
| 99 | + | // when recording until cancellation. | |||
| 100 | + | type schedule struct { | |||
| 101 | + | start time.Time | |||
| 102 | + | fps int64 | |||
| 103 | + | total int64 | |||
| 104 | + | } | |||
| 105 | + | ||||
| 106 | + | func newSchedule(start time.Time, sel ports.Selection) schedule { | |||
| 107 | + | sc := schedule{start: start, fps: int64(sel.Video.FPS)} | |||
| 108 | + | if sel.Duration > 0 { | |||
| 109 | + | // ceil(duration*fps) split into whole seconds and remainder so the | |||
| 110 | + | // arithmetic cannot overflow for any positive time.Duration. | |||
| 111 | + | secs, rem := int64(sel.Duration/time.Second), int64(sel.Duration%time.Second) | |||
| 112 | + | sc.total = secs*sc.fps + (rem*sc.fps+int64(time.Second)-1)/int64(time.Second) | |||
| 113 | + | } | |||
| 114 | + | return sc | |||
| 115 | + | } | |||
| 116 | + | ||||
| 117 | + | // due returns the offset at which slot n starts. | |||
| 118 | + | func (sc schedule) due(n int64) time.Duration { | |||
| 119 | + | q, r := n/sc.fps, n%sc.fps | |||
| 120 | + | return time.Duration(q)*time.Second + time.Duration(r)*time.Second/time.Duration(sc.fps) | |||
| 121 | + | } | |||
| 122 | + | ||||
| 123 | + | // current returns the slot containing now, limited to the last slot of a | |||
| 124 | + | // timed recording. past reports that a timed recording's window has ended. | |||
| 125 | + | func (sc schedule) current() (slot int64, past bool) { | |||
| 126 | + | elapsed := time.Since(sc.start) | |||
| 127 | + | secs, rem := int64(elapsed/time.Second), int64(elapsed%time.Second) | |||
| 128 | + | slot = secs*sc.fps + rem*sc.fps/int64(time.Second) | |||
| 129 | + | if sc.total > 0 && slot >= sc.total { | |||
| 130 | + | return sc.total - 1, true | |||
| 131 | + | } | |||
| 132 | + | return slot, false | |||
| 133 | + | } | |||
| 134 | + | ||||
| 135 | + | // recordLoop fills slots 1.. after the already written first frame. A slot | |||
| 136 | + | // missed because Capture or Write ran long reuses the current frame, which is | |||
| 137 | + | // still valid until the next Capture, so the encoded duration stays exact | |||
| 138 | + | // even when wall time overruns. A timed recording ends after ceil(duration*fps) | |||
| 139 | + | // frames; otherwise it ends when ctx is done. | |||
| 140 | + | func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.Selection) error { | |||
| 141 | + | sc := newSchedule(time.Now(), sel) | |||
| 142 | + | timer := time.NewTimer(time.Hour) | |||
| 143 | + | defer timer.Stop() | |||
| 144 | + | timer.Stop() | |||
| 145 | + | ||||
| 146 | + | next := int64(1) | |||
| 147 | + | for sc.total == 0 || next < sc.total { | |||
| 148 | + | if wait := sc.due(next) - time.Since(sc.start); wait > 0 { | |||
| 149 | + | timer.Reset(wait) | |||
| 150 | + | select { | |||
| 151 | + | case <-ctx.Done(): | |||
| 152 | + | return s.fillElapsed(ctx, frame, next, sc) | |||
| 153 | + | case <-timer.C: | |||
| 154 | + | } | |||
| 155 | + | } else if ctx.Err() != nil { | |||
| 156 | + | return s.fillElapsed(ctx, frame, next, sc) | |||
| 157 | + | } | |||
| 158 | + | ||||
| 159 | + | cur, past := sc.current() | |||
| 160 | + | cur = max(cur, next) | |||
| 161 | + | // Slots already missed repeat the frame before it is invalidated. | |||
| 162 | + | last := cur - 1 | |||
| 163 | + | if past { | |||
| 164 | + | last = cur // window over: no capture for the final slot | |||
| 165 | + | } | |||
| 166 | + | for ; next <= last; next++ { | |||
| 167 | + | if err := s.video.Write(ctx, frame); err != nil { | |||
| 168 | + | return writeResult(ctx, err) | |||
| 169 | + | } | |||
| 106 | } | 170 | } | ||
| 107 | - | if ctx.Err() != nil { | 171 | + | if past { |
| 108 | return nil | 172 | return nil | ||
| 109 | } | 173 | } | ||
| 110 | - | frame, err := s.source.Capture(ctx, sel.Target) | 174 | + | |
| 111 | - | if err != nil { | 175 | + | var err error |
| 176 | + | if frame, err = s.source.Capture(ctx, sel.Target); err != nil { | |||
| 112 | if stopped(ctx, err) { | 177 | if stopped(ctx, err) { | ||
| 113 | return nil | 178 | return nil | ||
| 114 | } | 179 | } | ||
| @@ -120,7 +185,26 @@ func (s *Service) recordLoop(ctx context.Context, first ports.Frame, sel ports.S | |||||
| 120 | if err := s.video.Write(ctx, frame); err != nil { | 185 | if err := s.video.Write(ctx, frame); err != nil { | ||
| 121 | return writeResult(ctx, err) | 186 | return writeResult(ctx, err) | ||
| 122 | } | 187 | } | ||
| 188 | + | next++ | |||
| 123 | } | 189 | } | ||
| 190 | + | return nil | |||
| 191 | + | } | |||
| 192 | + | ||||
| 193 | + | // fillElapsed repeats the current frame for slots that began before the stop | |||
| 194 | + | // request so the file covers the time actually recorded. | |||
| 195 | + | func (s *Service) fillElapsed(ctx context.Context, frame ports.Frame, next int64, sc schedule) error { | |||
| 196 | + | cur, _ := sc.current() | |||
| 197 | + | if next > cur { | |||
| 198 | + | return nil | |||
| 199 | + | } | |||
| 200 | + | wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), stopWriteTimeout) | |||
| 201 | + | defer cancel() | |||
| 202 | + | for ; next <= cur; next++ { | |||
| 203 | + | if err := s.video.Write(wctx, frame); err != nil { | |||
| 204 | + | return fmt.Errorf("write trailing frame: %w", err) | |||
| 205 | + | } | |||
| 206 | + | } | |||
| 207 | + | return nil | |||
| 124 | } | 208 | } | ||
| 125 | 209 | ||||
| 126 | func writeResult(ctx context.Context, err error) error { | 210 | func writeResult(ctx context.Context, err error) error { | ||
| @@ -130,9 +214,29 @@ func writeResult(ctx context.Context, err error) error { | |||||
| 130 | return fmt.Errorf("write frame: %w", err) | 214 | return fmt.Errorf("write frame: %w", err) | ||
| 131 | } | 215 | } | ||
| 132 | 216 | ||||
| 133 | - | // stopped reports whether err is only the consequence of the run ending. | 217 | + | // stopped reports whether the run is ending and err consists only of context |
| 218 | + | // cancellation or deadline leaves. A failure joined with cancellation is real. | |||
| 134 | func stopped(ctx context.Context, err error) bool { | 219 | func stopped(ctx context.Context, err error) bool { | ||
| 135 | - | return ctx.Err() != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) | 220 | + | return ctx.Err() != nil && onlyContext(err) |
| 221 | + | } | |||
| 222 | + | ||||
| 223 | + | func onlyContext(err error) bool { | |||
| 224 | + | switch e := err.(type) { | |||
| 225 | + | case nil: | |||
| 226 | + | return false | |||
| 227 | + | case interface{ Unwrap() []error }: | |||
| 228 | + | children := e.Unwrap() | |||
| 229 | + | for _, c := range children { | |||
| 230 | + | if !onlyContext(c) { | |||
| 231 | + | return false | |||
| 232 | + | } | |||
| 233 | + | } | |||
| 234 | + | return len(children) > 0 | |||
| 235 | + | case interface{ Unwrap() error }: | |||
| 236 | + | return err == context.Canceled || err == context.DeadlineExceeded || onlyContext(e.Unwrap()) | |||
| 237 | + | default: | |||
| 238 | + | return err == context.Canceled || err == context.DeadlineExceeded | |||
| 239 | + | } | |||
| 136 | } | 240 | } | ||
| 137 | 241 | ||||
| 138 | func validateSelection(sel ports.Selection) error { | 242 | func validateSelection(sel ports.Selection) error { | ||
M internal/core/service_test.go +174 -35
| @@ -3,6 +3,7 @@ package core_test | |||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | 5 | "errors" | |
| 6 | + | "fmt" | |
| 6 | 7 | "sync/atomic" | |
| 7 | 8 | "testing" | |
| 8 | 9 | "time" | |
| @@ -210,38 +211,6 @@ func recordOrder(h harness, sel ports.Selection) (first *mock.Call) { | |||
| 210 | 211 | return s | |
| 211 | 212 | } | |
| 212 | 213 | ||
| 213 | - | func TestRecordDuration(t *testing.T) { | |
| 214 | - | h := newHarness(t) | |
| 215 | - | sel := rec(60 * time.Millisecond) | |
| 216 | - | start := recordOrder(h, sel) | |
| 217 | - | var closed atomic.Bool | |
| 218 | - | var writes, inFlight atomic.Int32 | |
| 219 | - | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |
| 220 | - | require.False(t, closed.Load(), "capture after close") | |
| 221 | - | require.Equal(t, int32(1), inFlight.Add(1), "concurrent capture") | |
| 222 | - | defer inFlight.Add(-1) | |
| 223 | - | return frame, nil | |
| 224 | - | }).Maybe() | |
| 225 | - | w := h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error { | |
| 226 | - | require.False(t, closed.Load(), "write after close") | |
| 227 | - | require.Equal(t, int32(1), inFlight.Add(1), "concurrent write") | |
| 228 | - | defer inFlight.Add(-1) | |
| 229 | - | writes.Add(1) | |
| 230 | - | return nil | |
| 231 | - | }).NotBefore(start) | |
| 232 | - | w.Maybe() | |
| 233 | - | h.vid.EXPECT().Close().RunAndReturn(func() error { | |
| 234 | - | closed.Store(true) | |
| 235 | - | return nil | |
| 236 | - | }).Once() | |
| 237 | - | ||
| 238 | - | began := time.Now() | |
| 239 | - | require.NoError(t, h.svc.Run(context.Background(), sel)) | |
| 240 | - | assert.GreaterOrEqual(t, time.Since(began), 60*time.Millisecond) | |
| 241 | - | assert.GreaterOrEqual(t, writes.Load(), int32(2)) | |
| 242 | - | assert.True(t, closed.Load()) | |
| 243 | - | } | |
| 244 | - | ||
| 245 | 214 | func TestRecordFirstWriteCompletesBeforeTick(t *testing.T) { | |
| 246 | 215 | // 1 fps: only the first frame is written in a short run. | |
| 247 | 216 | h := newHarness(t) | |
| @@ -287,7 +256,7 @@ func TestRecordCancelInterruptsBlockedCapture(t *testing.T) { | |||
| 287 | 256 | require.NoError(t, h.svc.Run(ctx, sel)) | |
| 288 | 257 | } | |
| 289 | 258 | ||
| 290 | - | func TestRecordCancelDuringWrite(t *testing.T) { | |
| 259 | + | func TestRecordCancelDuringFirstWriteAborts(t *testing.T) { | |
| 291 | 260 | h := newHarness(t) | |
| 292 | 261 | sel := rec(0) | |
| 293 | 262 | ctx, cancel := context.WithCancel(context.Background()) | |
| @@ -296,7 +265,177 @@ func TestRecordCancelDuringWrite(t *testing.T) { | |||
| 296 | 265 | cancel() | |
| 297 | 266 | return ctx.Err() | |
| 298 | 267 | }).Once().NotBefore(start) | |
| 299 | - | h.vid.EXPECT().Close().Return(nil).Once().NotBefore(w) | |
| 268 | + | h.vid.EXPECT().Abort().Return(nil).Once().NotBefore(w) | |
| 269 | + | err := h.svc.Run(ctx, sel) | |
| 270 | + | require.ErrorIs(t, err, context.Canceled) | |
| 271 | + | } | |
| 272 | + | ||
| 273 | + | func TestRecordCancelDuringLaterWrite(t *testing.T) { | |
| 274 | + | h := newHarness(t) | |
| 275 | + | sel := rec(0) | |
| 276 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 277 | + | start := recordOrder(h, sel) | |
| 278 | + | first := h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |
| 279 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Once() | |
| 280 | + | h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(ctx context.Context, _ ports.Frame) error { | |
| 281 | + | cancel() | |
| 282 | + | return ctx.Err() | |
| 283 | + | }).Once().NotBefore(first) | |
| 284 | + | h.vid.EXPECT().Close().Return(nil).Once() | |
| 285 | + | require.NoError(t, h.svc.Run(ctx, sel)) | |
| 286 | + | } | |
| 287 | + | ||
| 288 | + | func TestRecordJoinedFailureIsNotSuppressedByCancellation(t *testing.T) { | |
| 289 | + | boom := errors.New("boom") | |
| 290 | + | joins := map[string]error{ | |
| 291 | + | "join": errors.Join(context.Canceled, boom), | |
| 292 | + | "wrapped": fmt.Errorf("x: %w", errors.Join(boom, fmt.Errorf("y: %w", context.Canceled))), | |
| 293 | + | "multi": fmt.Errorf("%w; %w", context.Canceled, boom), | |
| 294 | + | } | |
| 295 | + | for name, werr := range joins { | |
| 296 | + | t.Run("write/"+name, func(t *testing.T) { | |
| 297 | + | h := newHarness(t) | |
| 298 | + | sel := rec(0) | |
| 299 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 300 | + | start := recordOrder(h, sel) | |
| 301 | + | first := h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |
| 302 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Once() | |
| 303 | + | h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error { | |
| 304 | + | cancel() | |
| 305 | + | return werr | |
| 306 | + | }).Once().NotBefore(first) | |
| 307 | + | h.vid.EXPECT().Close().Return(nil).Once() | |
| 308 | + | require.ErrorIs(t, h.svc.Run(ctx, sel), boom) | |
| 309 | + | }) | |
| 310 | + | t.Run("capture/"+name, func(t *testing.T) { | |
| 311 | + | h := newHarness(t) | |
| 312 | + | sel := rec(0) | |
| 313 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 314 | + | start := recordOrder(h, sel) | |
| 315 | + | h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |
| 316 | + | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |
| 317 | + | cancel() | |
| 318 | + | return ports.Frame{}, werr | |
| 319 | + | }).Once() | |
| 320 | + | h.vid.EXPECT().Close().Return(nil).Once() | |
| 321 | + | require.ErrorIs(t, h.svc.Run(ctx, sel), boom) | |
| 322 | + | }) | |
| 323 | + | } | |
| 324 | + | t.Run("wrapped cancellation alone is a clean stop", func(t *testing.T) { | |
| 325 | + | h := newHarness(t) | |
| 326 | + | sel := rec(0) | |
| 327 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 328 | + | start := recordOrder(h, sel) | |
| 329 | + | h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |
| 330 | + | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |
| 331 | + | cancel() | |
| 332 | + | return ports.Frame{}, fmt.Errorf("wayland: %w", errors.Join(context.Canceled, context.Canceled)) | |
| 333 | + | }).Once() | |
| 334 | + | h.vid.EXPECT().Close().Return(nil).Once() | |
| 335 | + | require.NoError(t, h.svc.Run(ctx, sel)) | |
| 336 | + | }) | |
| 337 | + | } | |
| 338 | + | ||
| 339 | + | // slotRecorder tracks the most recently captured frame: only it may be written. | |
| 340 | + | type slotRecorder struct { | |
| 341 | + | current *byte | |
| 342 | + | captures atomic.Int32 | |
| 343 | + | writes atomic.Int32 | |
| 344 | + | } | |
| 345 | + | ||
| 346 | + | func newFrame() ports.Frame { | |
| 347 | + | f := frame | |
| 348 | + | f.Pixels = make([]byte, len(frame.Pixels)) | |
| 349 | + | return f | |
| 350 | + | } | |
| 351 | + | ||
| 352 | + | func TestRecordExactFrameCountWithSlowStartAndCapture(t *testing.T) { | |
| 353 | + | cases := []struct { | |
| 354 | + | name string | |
| 355 | + | duration time.Duration | |
| 356 | + | fps int | |
| 357 | + | want int32 | |
| 358 | + | }{ | |
| 359 | + | {"slow capture 60fps", 300 * time.Millisecond, 60, 18}, | |
| 360 | + | {"fractional slot rounds up", 25 * time.Millisecond, 60, 2}, | |
| 361 | + | {"single slot", 10 * time.Millisecond, 30, 1}, | |
| 362 | + | {"fast", 100 * time.Millisecond, 30, 3}, | |
| 363 | + | } | |
| 364 | + | for _, tc := range cases { | |
| 365 | + | t.Run(tc.name, func(t *testing.T) { | |
| 366 | + | h := newHarness(t) | |
| 367 | + | sel := rec(tc.duration) | |
| 368 | + | sel.Video.FPS = tc.fps | |
| 369 | + | var r slotRecorder | |
| 370 | + | first := newFrame() | |
| 371 | + | r.current = &first.Pixels[0] | |
| 372 | + | var closed atomic.Bool | |
| 373 | + | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |
| 374 | + | r.captures.Add(1) | |
| 375 | + | f := first | |
| 376 | + | if r.captures.Load() > 1 { | |
| 377 | + | time.Sleep(25 * time.Millisecond) | |
| 378 | + | f = newFrame() | |
| 379 | + | } | |
| 380 | + | r.current = &f.Pixels[0] | |
| 381 | + | return f, nil | |
| 382 | + | }).Maybe() | |
| 383 | + | s := h.vid.EXPECT().Start(mock.Anything, mock.Anything, sel.Path, sel.Video).RunAndReturn(func(context.Context, ports.Frame, string, ports.VideoSettings) error { | |
| 384 | + | time.Sleep(100 * time.Millisecond) // must not count against the duration | |
| 385 | + | return nil | |
| 386 | + | }).Once() | |
| 387 | + | h.vid.EXPECT().Write(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, f ports.Frame) error { | |
| 388 | + | require.False(t, closed.Load(), "write after close") | |
| 389 | + | require.Same(t, r.current, &f.Pixels[0], "stale or unknown frame written") | |
| 390 | + | r.writes.Add(1) | |
| 391 | + | return nil | |
| 392 | + | }).NotBefore(s).Maybe() | |
| 393 | + | h.vid.EXPECT().Close().RunAndReturn(func() error { | |
| 394 | + | closed.Store(true) | |
| 395 | + | return nil | |
| 396 | + | }).Once() | |
| 397 | + | ||
| 398 | + | began := time.Now() | |
| 399 | + | require.NoError(t, h.svc.Run(context.Background(), sel)) | |
| 400 | + | assert.Equal(t, tc.want, r.writes.Load()) | |
| 401 | + | assert.LessOrEqual(t, r.captures.Load(), tc.want) | |
| 402 | + | assert.GreaterOrEqual(t, time.Since(began), 100*time.Millisecond+tc.duration-time.Second/time.Duration(tc.fps)) | |
| 403 | + | }) | |
| 404 | + | } | |
| 405 | + | } | |
| 406 | + | ||
| 407 | + | func TestRecordStopWritesElapsedSlots(t *testing.T) { | |
| 408 | + | h := newHarness(t) | |
| 409 | + | sel := rec(0) | |
| 410 | + | sel.Video.FPS = 50 | |
| 411 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 412 | + | start := recordOrder(h, sel) | |
| 413 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Maybe() | |
| 414 | + | var writes atomic.Int32 | |
| 415 | + | h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error { | |
| 416 | + | writes.Add(1) | |
| 417 | + | return nil | |
| 418 | + | }).NotBefore(start).Maybe() | |
| 419 | + | h.vid.EXPECT().Close().Return(nil).Once() | |
| 420 | + | ||
| 421 | + | stop := time.AfterFunc(200*time.Millisecond, cancel) | |
| 422 | + | defer stop.Stop() | |
| 423 | + | began := time.Now() | |
| 424 | + | require.NoError(t, h.svc.Run(ctx, sel)) | |
| 425 | + | slots := int32(time.Since(began) * 50 / time.Second) | |
| 426 | + | assert.GreaterOrEqual(t, writes.Load(), int32(10)) | |
| 427 | + | assert.LessOrEqual(t, slots-writes.Load(), int32(1), "elapsed slots must be written") | |
| 428 | + | } | |
| 429 | + | ||
| 430 | + | func TestRecordHugeDurationDoesNotOverflow(t *testing.T) { | |
| 431 | + | h := newHarness(t) | |
| 432 | + | sel := rec(time.Duration(1<<63 - 1)) | |
| 433 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 434 | + | start := recordOrder(h, sel) | |
| 435 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Maybe() | |
| 436 | + | h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).NotBefore(start).Maybe() | |
| 437 | + | h.vid.EXPECT().Close().Return(nil).Once() | |
| 438 | + | time.AfterFunc(40*time.Millisecond, cancel) | |
| 300 | 439 | require.NoError(t, h.svc.Run(ctx, sel)) | |
| 301 | 440 | } | |
| 302 | 441 | ||
| @@ -349,7 +488,7 @@ func TestRecordRuntimeFailures(t *testing.T) { | |||
| 349 | 488 | sel := rec(0) | |
| 350 | 489 | start := recordOrder(h, sel) | |
| 351 | 490 | w := h.vid.EXPECT().Write(mock.Anything, frame).Return(boom).Once().NotBefore(start) | |
| 352 | - | h.vid.EXPECT().Close().Return(closeErr).Once().NotBefore(w) | |
| 491 | + | h.vid.EXPECT().Abort().Return(closeErr).Once().NotBefore(w) | |
| 353 | 492 | err := h.svc.Run(context.Background(), sel) | |
| 354 | 493 | require.ErrorIs(t, err, boom) | |
| 355 | 494 | require.ErrorIs(t, err, closeErr) | |
| @@ -3,6 +3,7 @@ package core_test | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | "errors" | 5 | "errors" | ||
| 6 | + | "fmt" | |||
| 6 | "sync/atomic" | 7 | "sync/atomic" | ||
| 7 | "testing" | 8 | "testing" | ||
| 8 | "time" | 9 | "time" | ||
| @@ -210,38 +211,6 @@ func recordOrder(h harness, sel ports.Selection) (first *mock.Call) { | |||||
| 210 | return s | 211 | return s | ||
| 211 | } | 212 | } | ||
| 212 | 213 | ||||
| 213 | - | func TestRecordDuration(t *testing.T) { | |||
| 214 | - | h := newHarness(t) | |||
| 215 | - | sel := rec(60 * time.Millisecond) | |||
| 216 | - | start := recordOrder(h, sel) | |||
| 217 | - | var closed atomic.Bool | |||
| 218 | - | var writes, inFlight atomic.Int32 | |||
| 219 | - | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |||
| 220 | - | require.False(t, closed.Load(), "capture after close") | |||
| 221 | - | require.Equal(t, int32(1), inFlight.Add(1), "concurrent capture") | |||
| 222 | - | defer inFlight.Add(-1) | |||
| 223 | - | return frame, nil | |||
| 224 | - | }).Maybe() | |||
| 225 | - | w := h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error { | |||
| 226 | - | require.False(t, closed.Load(), "write after close") | |||
| 227 | - | require.Equal(t, int32(1), inFlight.Add(1), "concurrent write") | |||
| 228 | - | defer inFlight.Add(-1) | |||
| 229 | - | writes.Add(1) | |||
| 230 | - | return nil | |||
| 231 | - | }).NotBefore(start) | |||
| 232 | - | w.Maybe() | |||
| 233 | - | h.vid.EXPECT().Close().RunAndReturn(func() error { | |||
| 234 | - | closed.Store(true) | |||
| 235 | - | return nil | |||
| 236 | - | }).Once() | |||
| 237 | - | ||||
| 238 | - | began := time.Now() | |||
| 239 | - | require.NoError(t, h.svc.Run(context.Background(), sel)) | |||
| 240 | - | assert.GreaterOrEqual(t, time.Since(began), 60*time.Millisecond) | |||
| 241 | - | assert.GreaterOrEqual(t, writes.Load(), int32(2)) | |||
| 242 | - | assert.True(t, closed.Load()) | |||
| 243 | - | } | |||
| 244 | - | ||||
| 245 | func TestRecordFirstWriteCompletesBeforeTick(t *testing.T) { | 214 | func TestRecordFirstWriteCompletesBeforeTick(t *testing.T) { | ||
| 246 | // 1 fps: only the first frame is written in a short run. | 215 | // 1 fps: only the first frame is written in a short run. | ||
| 247 | h := newHarness(t) | 216 | h := newHarness(t) | ||
| @@ -287,7 +256,7 @@ func TestRecordCancelInterruptsBlockedCapture(t *testing.T) { | |||||
| 287 | require.NoError(t, h.svc.Run(ctx, sel)) | 256 | require.NoError(t, h.svc.Run(ctx, sel)) | ||
| 288 | } | 257 | } | ||
| 289 | 258 | ||||
| 290 | - | func TestRecordCancelDuringWrite(t *testing.T) { | 259 | + | func TestRecordCancelDuringFirstWriteAborts(t *testing.T) { |
| 291 | h := newHarness(t) | 260 | h := newHarness(t) | ||
| 292 | sel := rec(0) | 261 | sel := rec(0) | ||
| 293 | ctx, cancel := context.WithCancel(context.Background()) | 262 | ctx, cancel := context.WithCancel(context.Background()) | ||
| @@ -296,7 +265,177 @@ func TestRecordCancelDuringWrite(t *testing.T) { | |||||
| 296 | cancel() | 265 | cancel() | ||
| 297 | return ctx.Err() | 266 | return ctx.Err() | ||
| 298 | }).Once().NotBefore(start) | 267 | }).Once().NotBefore(start) | ||
| 299 | - | h.vid.EXPECT().Close().Return(nil).Once().NotBefore(w) | 268 | + | h.vid.EXPECT().Abort().Return(nil).Once().NotBefore(w) |
| 269 | + | err := h.svc.Run(ctx, sel) | |||
| 270 | + | require.ErrorIs(t, err, context.Canceled) | |||
| 271 | + | } | |||
| 272 | + | ||||
| 273 | + | func TestRecordCancelDuringLaterWrite(t *testing.T) { | |||
| 274 | + | h := newHarness(t) | |||
| 275 | + | sel := rec(0) | |||
| 276 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 277 | + | start := recordOrder(h, sel) | |||
| 278 | + | first := h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |||
| 279 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Once() | |||
| 280 | + | h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(ctx context.Context, _ ports.Frame) error { | |||
| 281 | + | cancel() | |||
| 282 | + | return ctx.Err() | |||
| 283 | + | }).Once().NotBefore(first) | |||
| 284 | + | h.vid.EXPECT().Close().Return(nil).Once() | |||
| 285 | + | require.NoError(t, h.svc.Run(ctx, sel)) | |||
| 286 | + | } | |||
| 287 | + | ||||
| 288 | + | func TestRecordJoinedFailureIsNotSuppressedByCancellation(t *testing.T) { | |||
| 289 | + | boom := errors.New("boom") | |||
| 290 | + | joins := map[string]error{ | |||
| 291 | + | "join": errors.Join(context.Canceled, boom), | |||
| 292 | + | "wrapped": fmt.Errorf("x: %w", errors.Join(boom, fmt.Errorf("y: %w", context.Canceled))), | |||
| 293 | + | "multi": fmt.Errorf("%w; %w", context.Canceled, boom), | |||
| 294 | + | } | |||
| 295 | + | for name, werr := range joins { | |||
| 296 | + | t.Run("write/"+name, func(t *testing.T) { | |||
| 297 | + | h := newHarness(t) | |||
| 298 | + | sel := rec(0) | |||
| 299 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 300 | + | start := recordOrder(h, sel) | |||
| 301 | + | first := h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |||
| 302 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Once() | |||
| 303 | + | h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error { | |||
| 304 | + | cancel() | |||
| 305 | + | return werr | |||
| 306 | + | }).Once().NotBefore(first) | |||
| 307 | + | h.vid.EXPECT().Close().Return(nil).Once() | |||
| 308 | + | require.ErrorIs(t, h.svc.Run(ctx, sel), boom) | |||
| 309 | + | }) | |||
| 310 | + | t.Run("capture/"+name, func(t *testing.T) { | |||
| 311 | + | h := newHarness(t) | |||
| 312 | + | sel := rec(0) | |||
| 313 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 314 | + | start := recordOrder(h, sel) | |||
| 315 | + | h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |||
| 316 | + | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |||
| 317 | + | cancel() | |||
| 318 | + | return ports.Frame{}, werr | |||
| 319 | + | }).Once() | |||
| 320 | + | h.vid.EXPECT().Close().Return(nil).Once() | |||
| 321 | + | require.ErrorIs(t, h.svc.Run(ctx, sel), boom) | |||
| 322 | + | }) | |||
| 323 | + | } | |||
| 324 | + | t.Run("wrapped cancellation alone is a clean stop", func(t *testing.T) { | |||
| 325 | + | h := newHarness(t) | |||
| 326 | + | sel := rec(0) | |||
| 327 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 328 | + | start := recordOrder(h, sel) | |||
| 329 | + | h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).Once().NotBefore(start) | |||
| 330 | + | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |||
| 331 | + | cancel() | |||
| 332 | + | return ports.Frame{}, fmt.Errorf("wayland: %w", errors.Join(context.Canceled, context.Canceled)) | |||
| 333 | + | }).Once() | |||
| 334 | + | h.vid.EXPECT().Close().Return(nil).Once() | |||
| 335 | + | require.NoError(t, h.svc.Run(ctx, sel)) | |||
| 336 | + | }) | |||
| 337 | + | } | |||
| 338 | + | ||||
| 339 | + | // slotRecorder tracks the most recently captured frame: only it may be written. | |||
| 340 | + | type slotRecorder struct { | |||
| 341 | + | current *byte | |||
| 342 | + | captures atomic.Int32 | |||
| 343 | + | writes atomic.Int32 | |||
| 344 | + | } | |||
| 345 | + | ||||
| 346 | + | func newFrame() ports.Frame { | |||
| 347 | + | f := frame | |||
| 348 | + | f.Pixels = make([]byte, len(frame.Pixels)) | |||
| 349 | + | return f | |||
| 350 | + | } | |||
| 351 | + | ||||
| 352 | + | func TestRecordExactFrameCountWithSlowStartAndCapture(t *testing.T) { | |||
| 353 | + | cases := []struct { | |||
| 354 | + | name string | |||
| 355 | + | duration time.Duration | |||
| 356 | + | fps int | |||
| 357 | + | want int32 | |||
| 358 | + | }{ | |||
| 359 | + | {"slow capture 60fps", 300 * time.Millisecond, 60, 18}, | |||
| 360 | + | {"fractional slot rounds up", 25 * time.Millisecond, 60, 2}, | |||
| 361 | + | {"single slot", 10 * time.Millisecond, 30, 1}, | |||
| 362 | + | {"fast", 100 * time.Millisecond, 30, 3}, | |||
| 363 | + | } | |||
| 364 | + | for _, tc := range cases { | |||
| 365 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 366 | + | h := newHarness(t) | |||
| 367 | + | sel := rec(tc.duration) | |||
| 368 | + | sel.Video.FPS = tc.fps | |||
| 369 | + | var r slotRecorder | |||
| 370 | + | first := newFrame() | |||
| 371 | + | r.current = &first.Pixels[0] | |||
| 372 | + | var closed atomic.Bool | |||
| 373 | + | h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) { | |||
| 374 | + | r.captures.Add(1) | |||
| 375 | + | f := first | |||
| 376 | + | if r.captures.Load() > 1 { | |||
| 377 | + | time.Sleep(25 * time.Millisecond) | |||
| 378 | + | f = newFrame() | |||
| 379 | + | } | |||
| 380 | + | r.current = &f.Pixels[0] | |||
| 381 | + | return f, nil | |||
| 382 | + | }).Maybe() | |||
| 383 | + | s := h.vid.EXPECT().Start(mock.Anything, mock.Anything, sel.Path, sel.Video).RunAndReturn(func(context.Context, ports.Frame, string, ports.VideoSettings) error { | |||
| 384 | + | time.Sleep(100 * time.Millisecond) // must not count against the duration | |||
| 385 | + | return nil | |||
| 386 | + | }).Once() | |||
| 387 | + | h.vid.EXPECT().Write(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, f ports.Frame) error { | |||
| 388 | + | require.False(t, closed.Load(), "write after close") | |||
| 389 | + | require.Same(t, r.current, &f.Pixels[0], "stale or unknown frame written") | |||
| 390 | + | r.writes.Add(1) | |||
| 391 | + | return nil | |||
| 392 | + | }).NotBefore(s).Maybe() | |||
| 393 | + | h.vid.EXPECT().Close().RunAndReturn(func() error { | |||
| 394 | + | closed.Store(true) | |||
| 395 | + | return nil | |||
| 396 | + | }).Once() | |||
| 397 | + | ||||
| 398 | + | began := time.Now() | |||
| 399 | + | require.NoError(t, h.svc.Run(context.Background(), sel)) | |||
| 400 | + | assert.Equal(t, tc.want, r.writes.Load()) | |||
| 401 | + | assert.LessOrEqual(t, r.captures.Load(), tc.want) | |||
| 402 | + | assert.GreaterOrEqual(t, time.Since(began), 100*time.Millisecond+tc.duration-time.Second/time.Duration(tc.fps)) | |||
| 403 | + | }) | |||
| 404 | + | } | |||
| 405 | + | } | |||
| 406 | + | ||||
| 407 | + | func TestRecordStopWritesElapsedSlots(t *testing.T) { | |||
| 408 | + | h := newHarness(t) | |||
| 409 | + | sel := rec(0) | |||
| 410 | + | sel.Video.FPS = 50 | |||
| 411 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 412 | + | start := recordOrder(h, sel) | |||
| 413 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Maybe() | |||
| 414 | + | var writes atomic.Int32 | |||
| 415 | + | h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error { | |||
| 416 | + | writes.Add(1) | |||
| 417 | + | return nil | |||
| 418 | + | }).NotBefore(start).Maybe() | |||
| 419 | + | h.vid.EXPECT().Close().Return(nil).Once() | |||
| 420 | + | ||||
| 421 | + | stop := time.AfterFunc(200*time.Millisecond, cancel) | |||
| 422 | + | defer stop.Stop() | |||
| 423 | + | began := time.Now() | |||
| 424 | + | require.NoError(t, h.svc.Run(ctx, sel)) | |||
| 425 | + | slots := int32(time.Since(began) * 50 / time.Second) | |||
| 426 | + | assert.GreaterOrEqual(t, writes.Load(), int32(10)) | |||
| 427 | + | assert.LessOrEqual(t, slots-writes.Load(), int32(1), "elapsed slots must be written") | |||
| 428 | + | } | |||
| 429 | + | ||||
| 430 | + | func TestRecordHugeDurationDoesNotOverflow(t *testing.T) { | |||
| 431 | + | h := newHarness(t) | |||
| 432 | + | sel := rec(time.Duration(1<<63 - 1)) | |||
| 433 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 434 | + | start := recordOrder(h, sel) | |||
| 435 | + | h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Maybe() | |||
| 436 | + | h.vid.EXPECT().Write(mock.Anything, frame).Return(nil).NotBefore(start).Maybe() | |||
| 437 | + | h.vid.EXPECT().Close().Return(nil).Once() | |||
| 438 | + | time.AfterFunc(40*time.Millisecond, cancel) | |||
| 300 | require.NoError(t, h.svc.Run(ctx, sel)) | 439 | require.NoError(t, h.svc.Run(ctx, sel)) | ||
| 301 | } | 440 | } | ||
| 302 | 441 | ||||
| @@ -349,7 +488,7 @@ func TestRecordRuntimeFailures(t *testing.T) { | |||||
| 349 | sel := rec(0) | 488 | sel := rec(0) | ||
| 350 | start := recordOrder(h, sel) | 489 | start := recordOrder(h, sel) | ||
| 351 | w := h.vid.EXPECT().Write(mock.Anything, frame).Return(boom).Once().NotBefore(start) | 490 | w := h.vid.EXPECT().Write(mock.Anything, frame).Return(boom).Once().NotBefore(start) | ||
| 352 | - | h.vid.EXPECT().Close().Return(closeErr).Once().NotBefore(w) | 491 | + | h.vid.EXPECT().Abort().Return(closeErr).Once().NotBefore(w) |
| 353 | err := h.svc.Run(context.Background(), sel) | 492 | err := h.svc.Run(context.Background(), sel) | ||
| 354 | require.ErrorIs(t, err, boom) | 493 | require.ErrorIs(t, err, boom) | ||
| 355 | require.ErrorIs(t, err, closeErr) | 494 | require.ErrorIs(t, err, closeErr) | ||