diff --git a/cmd/skald-recording-smoke/main.go b/cmd/skald-recording-smoke/main.go new file mode 100644 index 0000000..97d4f20 --- /dev/null +++ b/cmd/skald-recording-smoke/main.go @@ -0,0 +1,35 @@ +package main + +import ( + "flag" + "fmt" + "log" + "time" + + "git.stormux.org/storm/skald/diskwriter" +) + +func main() { + var directory string + var sources int + var duration time.Duration + + flag.StringVar(&directory, "out", ".", "output directory") + flag.IntVar(&sources, "sources", 5, "synthetic active source count") + flag.DurationVar(&duration, "duration", 10*time.Second, "recording duration") + flag.Parse() + + result, err := diskwriter.RunSyntheticRecording(directory, sources, duration) + if err != nil { + log.Fatal(err) + } + + fmt.Printf("Recording: %s\n", result.Path) + fmt.Printf("Duration: %v\n", result.Metrics.Duration.Round(time.Millisecond)) + fmt.Printf("Decoder starts: %d\n", result.Metrics.DecoderStarts) + fmt.Printf("Decoder stops: %d\n", result.Metrics.DecoderStops) + fmt.Printf("Source frame drops: %d\n", result.Metrics.DroppedSourceFrames) + fmt.Printf("Stale source ticks: %d\n", result.Metrics.StaleSourceTicks) + fmt.Printf("Mixed frame drops: %d\n", result.Metrics.DroppedMixedFrames) + fmt.Printf("Encoder write errors: %d\n", result.Metrics.EncoderWriteErrors) +} diff --git a/diskwriter/diskwriter.go b/diskwriter/diskwriter.go index d34dff6..2b0c20e 100644 --- a/diskwriter/diskwriter.go +++ b/diskwriter/diskwriter.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "log" + "math" "net" "os" "os/exec" @@ -24,12 +25,16 @@ import ( ) const ( - sampleRate = 48000 - channels = 2 - frameDuration = 20 * time.Millisecond - samplesPerFrame = sampleRate / 1000 * 20 - bytesPerFrame = samplesPerFrame * channels * 2 - sourceBufferSize = 16 + sampleRate = 48000 + channels = 2 + frameDuration = 20 * time.Millisecond + samplesPerFrame = sampleRate / 1000 * 20 + bytesPerFrame = samplesPerFrame * channels * 2 + sourceBufferSize = 16 + encoderBufferSize = 50 + sourceStaleAfter = 2 * time.Second + limiterThreshold = 28000 + limiterRatio = 4 ) var Directory string @@ -197,17 +202,46 @@ func (client *Client) PushConn(g *hall.Hall, id string, up conn.Up, tracks []con } type recorder struct { - client *Client - path string - cmd *exec.Cmd - stdin io.WriteCloser - done chan error + client *Client + path string + cmd *exec.Cmd + stdin io.WriteCloser + done chan error + writeCh chan []byte + writerDone chan error mu sync.Mutex - sources map[*diskTrack]chan []byte + sources map[*diskTrack]*recordingSource + metrics RecorderMetrics closed bool } +type recordingSource struct { + ch chan []byte + lastFrame time.Time + decoderStarted bool + decoderStopped bool + staleLogged bool +} + +type RecorderMetrics struct { + Started time.Time + Duration time.Duration + ActiveSources int + DecoderStarts uint64 + DecoderStops uint64 + DroppedSourceFrames uint64 + StaleSourceTicks uint64 + DroppedMixedFrames uint64 + EncoderWriteErrors uint64 + MixerFrames uint64 +} + +type SyntheticRecordingResult struct { + Path string + Metrics RecorderMetrics +} + func newRecorder(client *Client, directory string) (*recorder, error) { file, err := openDiskFile(directory, "ogg") if err != nil { @@ -245,35 +279,107 @@ func newRecorder(client *Client, directory string) (*recorder, error) { go drainStderr("ffmpeg encoder", stderr) r := &recorder{ - client: client, - path: path, - cmd: cmd, - stdin: stdin, - done: make(chan error, 1), - sources: make(map[*diskTrack]chan []byte), + client: client, + path: path, + cmd: cmd, + stdin: stdin, + done: make(chan error, 1), + writeCh: make(chan []byte, encoderBufferSize), + writerDone: make(chan error, 1), + sources: make(map[*diskTrack]*recordingSource), + metrics: RecorderMetrics{Started: time.Now()}, } activeRecordings.Store(path, struct{}{}) + go r.writeEncoder() go r.mix() return r, nil } -func (r *recorder) addSource(t *diskTrack) chan []byte { +func (r *recorder) addSource(t *diskTrack) *recordingSource { r.mu.Lock() defer r.mu.Unlock() if r.closed { return nil } - ch := make(chan []byte, sourceBufferSize) - r.sources[t] = ch - return ch + source := &recordingSource{ + ch: make(chan []byte, sourceBufferSize), + lastFrame: time.Now(), + } + r.sources[t] = source + return source +} + +func (r *recorder) noteDecoderStarted(t *diskTrack) { + r.mu.Lock() + defer r.mu.Unlock() + + source := r.sources[t] + if source == nil || source.decoderStarted { + return + } + source.decoderStarted = true + r.metrics.DecoderStarts++ + log.Printf("Recording %s: decoder started, active sources %d", + filepath.Base(r.path), len(r.sources)) } func (r *recorder) removeSource(t *diskTrack) { r.mu.Lock() defer r.mu.Unlock() + source := r.sources[t] + if source == nil { + return + } + if source.decoderStarted && !source.decoderStopped { + source.decoderStopped = true + r.metrics.DecoderStops++ + } delete(r.sources, t) + log.Printf("Recording %s: source removed, active sources %d", + filepath.Base(r.path), len(r.sources)) +} + +func (r *recorder) noteDecoderStopped(t *diskTrack, err error) { + r.mu.Lock() + defer r.mu.Unlock() + + source := r.sources[t] + if source == nil || source.decoderStopped { + return + } + source.decoderStopped = true + r.metrics.DecoderStops++ + delete(r.sources, t) + if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) { + log.Printf("Recording %s: decoder stopped: %v", + filepath.Base(r.path), err) + } else { + log.Printf("Recording %s: decoder stopped, active sources %d", + filepath.Base(r.path), len(r.sources)) + } +} + +func (r *recorder) pushSourceFrame(t *diskTrack, frame []byte) bool { + r.mu.Lock() + defer r.mu.Unlock() + + source := r.sources[t] + if source == nil { + return false + } + now := time.Now() + source.lastFrame = now + source.staleLogged = false + select { + case source.ch <- frame: + default: + <-source.ch + source.ch <- frame + r.metrics.DroppedSourceFrames++ + } + return true } func (r *recorder) mix() { @@ -281,74 +387,149 @@ func (r *recorder) mix() { defer ticker.Stop() silence := make([]byte, bytesPerFrame) - for range ticker.C { - r.mu.Lock() - if r.closed { - r.mu.Unlock() - r.done <- nil - return - } - sources := make([]chan []byte, 0, len(r.sources)) - for _, ch := range r.sources { - sources = append(sources, ch) - } - r.mu.Unlock() - - frame := mixFrame(sources) - if frame == nil { - frame = silence - } - if _, err := r.stdin.Write(frame); err != nil { + for { + select { + case err := <-r.writerDone: + if err != nil { + r.mu.Lock() + r.metrics.EncoderWriteErrors++ + r.mu.Unlock() + } r.done <- err return + case <-ticker.C: + if r.isClosed() { + close(r.writeCh) + err := <-r.writerDone + if err != nil { + r.mu.Lock() + r.metrics.EncoderWriteErrors++ + r.mu.Unlock() + } + r.done <- err + return + } + frame := r.nextMixedFrame(time.Now()) + if frame == nil { + frame = silence + } + r.queueMixedFrame(frame) } } } -func mixFrame(sources []chan []byte) []byte { - var mixed []int32 - used := false +func (r *recorder) isClosed() bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.closed +} - for _, ch := range sources { - var latest []byte - for { - select { - case b := <-ch: - latest = b - default: - goto drained +func (r *recorder) nextMixedFrame(now time.Time) []byte { + r.mu.Lock() + frames := make([][]byte, 0, len(r.sources)) + for _, source := range r.sources { + select { + case frame := <-source.ch: + if len(frame) >= bytesPerFrame { + frames = append(frames, frame) + } + default: + if now.Sub(source.lastFrame) >= sourceStaleAfter && !source.staleLogged { + source.staleLogged = true + r.metrics.StaleSourceTicks++ + log.Printf("Recording %s: source has no decoded frames for %v", + filepath.Base(r.path), sourceStaleAfter) } } - drained: - if len(latest) < bytesPerFrame { - continue + } + r.metrics.MixerFrames++ + r.mu.Unlock() + + return mixPCMFrames(frames) +} + +func (r *recorder) queueMixedFrame(frame []byte) { + select { + case r.writeCh <- frame: + default: + r.mu.Lock() + r.metrics.DroppedMixedFrames++ + r.mu.Unlock() + log.Printf("Recording %s: encoder queue full, dropping mixed frame", + filepath.Base(r.path)) + } +} + +func (r *recorder) writeEncoder() { + for frame := range r.writeCh { + if _, err := r.stdin.Write(frame); err != nil { + r.writerDone <- err + return } + } + r.writerDone <- nil +} + +func mixFrame(sources []chan []byte) []byte { + frames := make([][]byte, 0, len(sources)) + for _, ch := range sources { + var frame []byte + select { + case frame = <-ch: + default: + } + if len(frame) >= bytesPerFrame { + frames = append(frames, frame) + } + } + return mixPCMFrames(frames) +} + +func mixPCMFrames(frames [][]byte) []byte { + if len(frames) == 0 { + return nil + } + + var mixed []int32 + gain := 1.0 + if len(frames) > 1 { + gain = 1 / math.Sqrt(float64(len(frames))) + } + + for _, frame := range frames { if mixed == nil { mixed = make([]int32, samplesPerFrame*channels) } for i := range mixed { - sample := int16(binary.LittleEndian.Uint16(latest[i*2:])) + sample := int16(binary.LittleEndian.Uint16(frame[i*2:])) mixed[i] += int32(sample) } - used = true - } - - if !used { - return nil } out := make([]byte, bytesPerFrame) for i, sample := range mixed { - if sample > 32767 { - sample = 32767 - } else if sample < -32768 { - sample = -32768 - } + sample = int32(math.Round(float64(sample) * gain)) + sample = limitSample(sample) binary.LittleEndian.PutUint16(out[i*2:], uint16(int16(sample))) } return out } +func limitSample(sample int32) int32 { + if sample > limiterThreshold { + sample = limiterThreshold + (sample-limiterThreshold)/limiterRatio + } else if sample < -limiterThreshold { + sample = -limiterThreshold + (sample+limiterThreshold)/limiterRatio + } + if sample > 32767 { + return 32767 + } + if sample < -32768 { + return -32768 + } + return sample +} + func (r *recorder) Close() error { r.mu.Lock() if r.closed { @@ -362,6 +543,18 @@ func (r *recorder) Close() error { stdinErr := r.stdin.Close() waitErr := r.cmd.Wait() activeRecordings.Delete(r.path) + metrics := r.snapshotMetrics() + log.Printf("Recording %s: closed after %v, sources=%d, decoders=%d/%d, source drops=%d, stale ticks=%d, mixed drops=%d, encoder errors=%d", + filepath.Base(r.path), + metrics.Duration, + metrics.ActiveSources, + metrics.DecoderStarts, + metrics.DecoderStops, + metrics.DroppedSourceFrames, + metrics.StaleSourceTicks, + metrics.DroppedMixedFrames, + metrics.EncoderWriteErrors, + ) if err != nil { return err @@ -379,6 +572,91 @@ func (r *recorder) Close() error { return nil } +func (r *recorder) snapshotMetrics() RecorderMetrics { + r.mu.Lock() + defer r.mu.Unlock() + metrics := r.metrics + metrics.Duration = time.Since(metrics.Started) + metrics.ActiveSources = len(r.sources) + return metrics +} + +func RunSyntheticRecording(directory string, sources int, duration time.Duration) (SyntheticRecordingResult, error) { + if sources <= 0 { + return SyntheticRecordingResult{}, errors.New("sources must be greater than zero") + } + if duration <= 0 { + return SyntheticRecordingResult{}, errors.New("duration must be greater than zero") + } + if err := os.MkdirAll(directory, 0700); err != nil { + return SyntheticRecordingResult{}, err + } + + r, err := newRecorder(nil, directory) + if err != nil { + return SyntheticRecordingResult{}, err + } + + var wg sync.WaitGroup + for i := 0; i < sources; i++ { + track := &diskTrack{} + if r.addSource(track) == nil { + r.Close() + return SyntheticRecordingResult{}, errors.New("recorder is closed") + } + r.noteDecoderStarted(track) + wg.Add(1) + go func(index int, t *diskTrack) { + defer wg.Done() + defer r.removeSource(t) + + ticker := time.NewTicker(frameDuration) + defer ticker.Stop() + timer := time.NewTimer(duration) + defer timer.Stop() + + frequency := 220 + float64(index)*37 + phase := 0.0 + for { + select { + case <-timer.C: + return + case <-ticker.C: + r.pushSourceFrame(t, sinePCMFrame(frequency, &phase)) + } + } + }(i, track) + } + wg.Wait() + + if err := r.Close(); err != nil { + return SyntheticRecordingResult{}, err + } + return SyntheticRecordingResult{ + Path: r.path, + Metrics: r.snapshotMetrics(), + }, nil +} + +func sinePCMFrame(frequency float64, phase *float64) []byte { + frame := make([]byte, bytesPerFrame) + step := 2 * math.Pi * frequency / sampleRate + for i := 0; i < samplesPerFrame; i++ { + sample := int16(math.Sin(*phase) * 6000) + *phase += step + if *phase >= 2*math.Pi { + *phase -= 2 * math.Pi + } + offset := i * channels * 2 + for ch := 0; ch < channels; ch++ { + binary.LittleEndian.PutUint16( + frame[offset+ch*2:], uint16(sample), + ) + } + } + return frame +} + type diskConn struct { client *Client remote conn.Up @@ -486,8 +764,8 @@ func (t *diskTrack) Write(buf []byte) (int, error) { } func (t *diskTrack) start(payload uint8) error { - ch := t.conn.client.recorder.addSource(t) - if ch == nil { + source := t.conn.client.recorder.addSource(t) + if source == nil { return errors.New("recorder is closed") } @@ -560,25 +838,26 @@ func (t *diskTrack) start(payload uint8) error { t.rtpConn = rtpConn t.cmd = cmd t.stdin = stdin - t.ch = ch + t.ch = source.ch + t.conn.client.recorder.noteDecoderStarted(t) go drainStderr("ffmpeg decoder", stderr) - go t.readPCM(stdout, ch) + go t.readPCM(stdout) return nil } -func (t *diskTrack) readPCM(r io.Reader, ch chan []byte) { +func (t *diskTrack) readPCM(r io.Reader) { for { buf := make([]byte, bytesPerFrame) _, err := io.ReadFull(r, buf) if err != nil { + if t.conn.client.recorder != nil { + t.conn.client.recorder.noteDecoderStopped(t, err) + } return } - select { - case ch <- buf: - default: - <-ch - ch <- buf + if t.conn.client.recorder != nil { + t.conn.client.recorder.pushSourceFrame(t, buf) } } } diff --git a/diskwriter/diskwriter_test.go b/diskwriter/diskwriter_test.go index ef69512..62ad411 100644 --- a/diskwriter/diskwriter_test.go +++ b/diskwriter/diskwriter_test.go @@ -103,7 +103,7 @@ func TestCleanupOldRecordings(t *testing.T) { } } -func TestMixFrameClipsAndSumsSources(t *testing.T) { +func TestMixFrameAppliesHeadroomForMultipleSources(t *testing.T) { a := make(chan []byte, 1) b := make(chan []byte, 1) a <- pcmFrame(20000) @@ -116,12 +116,79 @@ func TestMixFrameClipsAndSumsSources(t *testing.T) { for i := 0; i < len(out); i += 2 { got := int16(binary.LittleEndian.Uint16(out[i:])) - if got != 32767 { - t.Fatalf("sample %d = %d, expected clipped 32767", i/2, got) + if got != 28071 { + t.Fatalf("sample %d = %d, expected headroom-adjusted 28071", i/2, got) } } } +func TestMixFrameLimiterPreservesNormalSpeechLevel(t *testing.T) { + a := make(chan []byte, 1) + a <- pcmFrame(12000) + + out := mixFrame([]chan []byte{a}) + for i := 0; i < len(out); i += 2 { + got := int16(binary.LittleEndian.Uint16(out[i:])) + if got != 12000 { + t.Fatalf("sample %d = %d, expected unchanged 12000", i/2, got) + } + } +} + +func TestMixFrameHandlesQuietAndLoudSources(t *testing.T) { + a := make(chan []byte, 1) + b := make(chan []byte, 1) + a <- pcmFrame(1000) + b <- pcmFrame(20000) + + out := mixFrame([]chan []byte{a, b}) + for i := 0; i < len(out); i += 2 { + got := int16(binary.LittleEndian.Uint16(out[i:])) + if got != 14849 { + t.Fatalf("sample %d = %d, expected mixed quiet/loud sample 14849", i/2, got) + } + } +} + +func TestMixFrameLimiterBoundsOverloadedSamples(t *testing.T) { + a := make(chan []byte, 1) + b := make(chan []byte, 1) + a <- pcmFrame(32000) + b <- pcmFrame(32000) + + out := mixFrame([]chan []byte{a, b}) + for i := 0; i < len(out); i += 2 { + got := int16(binary.LittleEndian.Uint16(out[i:])) + if got <= 0 || got > 32767 { + t.Fatalf("sample %d = %d, expected bounded positive sample", i/2, got) + } + if got >= 32767 { + t.Fatalf("sample %d = %d, expected limiter below hard clip", i/2, got) + } + } +} + +func TestMixFrameConsumesOldestQueuedFrame(t *testing.T) { + source := make(chan []byte, 2) + source <- pcmFrame(1000) + source <- pcmFrame(2000) + + out := mixFrame([]chan []byte{source}) + if len(out) != bytesPerFrame { + t.Fatalf("mixed frame has length %d, expected %d", len(out), bytesPerFrame) + } + + for i := 0; i < len(out); i += 2 { + got := int16(binary.LittleEndian.Uint16(out[i:])) + if got != 1000 { + t.Fatalf("sample %d = %d, expected oldest queued sample 1000", i/2, got) + } + } + if queued := len(source); queued != 1 { + t.Fatalf("queued frames after mix = %d, expected 1", queued) + } +} + func TestMixFrameReturnsNilForSilence(t *testing.T) { if out := mixFrame(nil); out != nil { t.Fatalf("expected nil silent frame, got %d bytes", len(out)) @@ -162,6 +229,45 @@ func TestRecorderCreatesPlayableOgg(t *testing.T) { } } +func TestSyntheticRecordingCreatesPlayableOgg(t *testing.T) { + if _, err := exec.LookPath("ffmpeg"); err != nil { + t.Skip("ffmpeg not installed") + } + if _, err := exec.LookPath("ffprobe"); err != nil { + t.Skip("ffprobe not installed") + } + + result, err := RunSyntheticRecording(t.TempDir(), 3, 100*time.Millisecond) + if err != nil { + t.Fatal(err) + } + if result.Metrics.DecoderStarts != 3 { + t.Fatalf("decoder starts = %d, expected 3", result.Metrics.DecoderStarts) + } + if result.Metrics.DecoderStops != 3 { + t.Fatalf("decoder stops = %d, expected 3", result.Metrics.DecoderStops) + } + if result.Metrics.DroppedMixedFrames != 0 { + t.Fatalf("dropped mixed frames = %d, expected 0", result.Metrics.DroppedMixedFrames) + } + + cmd := exec.Command( + "ffprobe", + "-v", "error", + "-select_streams", "a:0", + "-show_entries", "stream=codec_name", + "-of", "default=noprint_wrappers=1:nokey=1", + result.Path, + ) + out, err := cmd.Output() + if err != nil { + t.Fatal(err) + } + if strings.TrimSpace(string(out)) != "opus" { + t.Fatalf("expected opus stream, got %q", strings.TrimSpace(string(out))) + } +} + func pcmFrame(sample int16) []byte { out := make([]byte, bytesPerFrame) for i := 0; i < len(out); i += 2 { diff --git a/distro-packages/Arch-Linux/skald-git/PKGBUILD b/distro-packages/Arch-Linux/skald-git/PKGBUILD index 38ce667..9af69ec 100644 --- a/distro-packages/Arch-Linux/skald-git/PKGBUILD +++ b/distro-packages/Arch-Linux/skald-git/PKGBUILD @@ -2,7 +2,7 @@ # shellcheck shell=bash disable=SC2034,SC2154 pkgname=skald-git -pkgver=0.0.0.r1530.g2e28dff +pkgver=0.0.0.r1531.g08b484f pkgrel=1 pkgdesc='Audio-only hall-based conferencing server' arch=('x86_64' 'aarch64') diff --git a/hall/hall.go b/hall/hall.go index 61e826e..8c7cf97 100644 --- a/hall/hall.go +++ b/hall/hall.go @@ -174,7 +174,17 @@ func (g *Hall) Description() *Description { func (g *Hall) ClientCount() int { g.mu.Lock() defer g.mu.Unlock() - return len(g.clients) + return g.clientCountUnlocked() +} + +func (g *Hall) clientCountUnlocked() int { + count := 0 + for _, c := range g.clients { + if !isSystemClient(c) { + count++ + } + } + return count } func (g *Hall) mayExpire() bool { @@ -525,6 +535,10 @@ func member(v string, l []string) bool { return false } +func isSystemClient(c Client) bool { + return member("system", c.Permissions()) +} + func AddClient(hall string, c Client, creds ClientCredentials) (*Hall, error) { g, err := Add(hall, nil) if err != nil { @@ -535,8 +549,9 @@ func AddClient(hall string, c Client, creds ClientCredentials) (*Hall, error) { defer g.mu.Unlock() clients := g.getClientsUnlocked(nil) + systemClient := isSystemClient(c) - if !member("system", c.Permissions()) { + if !systemClient { username, perms, err := g.getPermission(creds) if err != nil { return nil, err @@ -587,7 +602,7 @@ func AddClient(hall string, c Client, creds ClientCredentials) (*Hall, error) { } if !member("op", perms) && g.description.MaxClients > 0 { - if len(g.clients) >= g.description.MaxClients { + if g.clientCountUnlocked() >= g.description.MaxClients { return nil, UserError("too many users") } } @@ -607,12 +622,20 @@ func AddClient(hall string, c Client, creds ClientCredentials) (*Hall, error) { u := c.Username() p := c.Permissions() s := c.Data() - c.PushClient(g.Name(), "add", c.Id(), u, p, s) + if !systemClient { + c.PushClient(g.Name(), "add", c.Id(), u, p, s) + } for _, cc := range clients { pp := cc.Permissions() uu := cc.Username() - c.PushClient(g.Name(), "add", cc.Id(), uu, pp, cc.Data()) - cc.PushClient(g.Name(), "add", id, u, p, s) + if !isSystemClient(cc) { + c.PushClient( + g.Name(), "add", cc.Id(), uu, pp, cc.Data(), + ) + } + if !systemClient { + cc.PushClient(g.Name(), "add", id, u, p, s) + } } return g, nil @@ -659,6 +682,7 @@ func DelClient(c Client) { if g == nil { return } + systemClient := isSystemClient(c) g.mu.Lock() if g.clients[c.Id()] != c { log.Printf("Deleting unknown client") @@ -671,10 +695,12 @@ func DelClient(c Client) { g.mu.Unlock() c.Joined(g.Name(), "leave") - for _, cc := range clients { - cc.PushClient( - g.Name(), "delete", c.Id(), c.Username(), nil, nil, - ) + if !systemClient { + for _, cc := range clients { + cc.PushClient( + g.Name(), "delete", c.Id(), c.Username(), nil, nil, + ) + } } closeSystemClientsIfNoUsers(g, clients) autoLockKick(g) diff --git a/hall/hall_test.go b/hall/hall_test.go index 85cd10b..223d2ec 100644 --- a/hall/hall_test.go +++ b/hall/hall_test.go @@ -129,6 +129,14 @@ type testClient struct { id string username string permissions []string + pushed []pushedClient +} + +type pushedClient struct { + kind string + id string + username string + permissions []string } func (c *testClient) Hall() *Hall { return nil } @@ -144,11 +152,92 @@ func (c *testClient) PushConn(*Hall, string, conn.Up, []conn.UpTrack, string) er } func (c *testClient) RequestConns(Client, *Hall, string) error { return nil } func (c *testClient) Joined(string, string) error { return nil } -func (c *testClient) PushClient(string, string, string, string, []string, map[string]interface{}) error { +func (c *testClient) PushClient(_ string, kind string, id string, username string, permissions []string, _ map[string]interface{}) error { + c.pushed = append(c.pushed, pushedClient{ + kind: kind, + id: id, + username: username, + permissions: permissions, + }) return nil } func (c *testClient) Kick(string, *string, string) error { return nil } +func TestSystemClientIsNotAdvertisedToUsers(t *testing.T) { + halls.halls = nil + oldDirectory := Directory + Directory = t.TempDir() + t.Cleanup(func() { + Directory = oldDirectory + halls.halls = nil + }) + + err := os.WriteFile( + filepath.Join(Directory, "system-hidden-test.json"), + []byte(`{ + "users": { + "user1": {"password": "secret", "permissions": "present"}, + "user2": {"password": "secret", "permissions": "present"} + } + }`), + 0600, + ) + if err != nil { + t.Fatal(err) + } + + user1 := &testClient{id: "user1-client"} + g, err := AddClient("system-hidden-test", user1, ClientCredentials{ + Username: strPtr("user1"), + Password: "secret", + }) + if err != nil { + t.Fatal(err) + } + user1.pushed = nil + + recorder := &testClient{ + id: "recorder-client", + username: "RECORDING", + permissions: []string{"system"}, + } + _, err = AddClient("system-hidden-test", recorder, ClientCredentials{ + System: true, + }) + if err != nil { + t.Fatal(err) + } + if len(user1.pushed) != 0 { + t.Fatalf("user saw system recorder client: %#v", user1.pushed) + } + if count := g.ClientCount(); count != 1 { + t.Fatalf("client count with recorder = %d, expected 1", count) + } + + user2 := &testClient{id: "user2-client"} + _, err = AddClient("system-hidden-test", user2, ClientCredentials{ + Username: strPtr("user2"), + Password: "secret", + }) + if err != nil { + t.Fatal(err) + } + for _, pushed := range user2.pushed { + if pushed.id == recorder.id { + t.Fatalf("joining user saw system recorder client: %#v", user2.pushed) + } + } + for _, pushed := range user1.pushed { + if pushed.id == recorder.id { + t.Fatalf("existing user saw system recorder client: %#v", user1.pushed) + } + } +} + +func strPtr(s string) *string { + return &s +} + func TestLockedHallJoinMessage(t *testing.T) { halls.halls = nil oldDirectory := Directory diff --git a/skald.md b/skald.md index f7f64ff..4bc306d 100644 --- a/skald.md +++ b/skald.md @@ -423,12 +423,32 @@ recordings directory. The file is named `skald-recording-YYYYMMDD-HHMMSS.ogg` and contains a single mixed Ogg Opus audio stream. -Skald uses `ffmpeg` to decode incoming Opus RTP audio and encode the mixed -output. If `ffmpeg` is not installed or cannot be started, recording fails -with an operator-visible error instead of silently creating a broken file. +Skald uses one `ffmpeg` decoder process for each active incoming Opus RTP +audio source, mixes decoded 48 kHz stereo PCM every 20 ms, and sends the +mixed stream to one `ffmpeg` encoder process that writes the Ogg Opus file. +The mixer treats missing source frames as silence, applies deterministic +headroom and limiting before encoding, and logs recording metrics when the +file closes: active sources, decoder starts and stops, dropped source frames, +stale source ticks, dropped mixed frames, and encoder write errors. + +The encoder is fed through a bounded queue so a stuck encoder cannot block +WebRTC packet handling indefinitely. If the queue fills, Skald drops mixed +frames and logs the condition. If `ffmpeg` is not installed or cannot be +started, recording fails with an operator-visible error instead of silently +creating a broken file. The recorder closes when recording is stopped explicitly or when the hall no longer has any non-system clients. +Maintainers can exercise the recording mixer and encoder without browser +clients by running: + +``` +go run ./cmd/skald-recording-smoke -out /tmp -sources 5 -duration 10s +``` + +This creates a real mixed Ogg Opus file from synthetic PCM sources and prints +the same core counters that are useful when checking larger halls. + If `recordingRetention` is set in `data/config.json`, Skald periodically deletes completed recording files older than the configured duration. diff --git a/todo.txt b/todo.txt index 3ab38fd..ef31b96 100644 --- a/todo.txt +++ b/todo.txt @@ -28,14 +28,16 @@ X Confirm the current live deployment can start and stop a Skald recording. X Confirm migrated Galene rooms work as Skald halls on meet.stormux.org. -- Document the current recording pipeline in skald.md for maintainers: +X Document the current recording pipeline in skald.md for maintainers: incoming Opus RTP -> per-source ffmpeg decoder -> PCM mixer -> ffmpeg Opus encoder -> Ogg Opus file. -- Add lightweight recording metrics in logs or debug output: +X Add lightweight recording metrics in logs or debug output: active recorded sources, dropped source frames, mixer write errors, ffmpeg decoder starts/stops, and encoder failures. -- Add a synthetic recording benchmark or smoke tool that can simulate multiple - active Opus sources without needing browser clients. +X Add a synthetic recording smoke tool that can simulate multiple decoded PCM + sources without needing browser clients. +- Add a synthetic RTP/Opus recording benchmark that exercises per-source + decoder subprocesses without needing browser clients. - Record baseline CPU, process count, and output quality for 1, 5, 10, and 25 active sources on a representative server. - Define practical target limits for "normal hall", "large moderated hall", @@ -45,13 +47,13 @@ X Confirm migrated Galene rooms work as Skald halls on meet.stormux.org. # PHASE 1: MIX QUALITY BEFORE PIPELINE REWRITE # ============================================================================ -- Add mixer tests for multiple simultaneous PCM sources, including clipping +X Add mixer tests for multiple simultaneous PCM sources, including clipping cases and quiet/loud source combinations. -- Replace raw summing with a safer mix strategy: +X Replace raw summing with a safer mix strategy: per-source attenuation, headroom, or another deterministic approach that reduces clipping when multiple users speak. -- Add a simple limiter to prevent harsh clipping in the final mixed PCM. -- Add tests proving the limiter preserves normal speech-level samples and +X Add a simple limiter to prevent harsh clipping in the final mixed PCM. +X Add tests proving the limiter preserves normal speech-level samples and bounds overloaded samples. - Add optional per-source level tracking so logs or future UI can identify sources that dominate the recording. @@ -64,8 +66,8 @@ X Confirm migrated Galene rooms work as Skald halls on meet.stormux.org. # PHASE 2: SOURCE LIFECYCLE AND SILENCE HANDLING # ============================================================================ -- Track whether each recording source is active, silent, stalled, or closed. -- Treat missing frames as silence without blocking the mixer. +X Track whether each recording source is active, silent, stalled, or closed. +X Treat missing frames as silence without blocking the mixer. - Drop or de-prioritize stale silent sources after a short timeout. - Ensure mute/unmute transitions do not leave stale audio in the mixed output. - Ensure participant join/leave during recording never stops the recording file. @@ -99,12 +101,12 @@ X Confirm migrated Galene rooms work as Skald halls on meet.stormux.org. - Decide whether the final Ogg Opus encoder should remain ffmpeg or move in-process later. - Make encoder startup errors operator-visible and log the exact ffmpeg error. -- Ensure encoder stdin writes cannot block the mixer indefinitely. +X Ensure encoder stdin writes cannot block the mixer indefinitely. - Ensure recorder close handles encoder failure, stdin close failure, and wait failure without leaving misleading successful output. - Verify interrupted or failed recordings are either clearly marked as failed or produce a playable partial file. -- Verify completed files with ffprobe or equivalent in tests where available. +X Verify completed files with ffprobe or equivalent in tests where available. - Verify completed files play in common players after long recordings. # ============================================================================