Attempt to make audio and speech discovery more robust.

This commit is contained in:
Storm Dragon committed 2026-11-03 13:21:50 -05:00
1 parent a27f274099
commit dc945ff392
21 files changed
+1314 -60

No files matched your search

+9
View File
@@ -1,5 +1,14 @@
# Repository Guidelines # Repository Guidelines
## CRITICAL: Init-System Independence
Fenrir must never depend on a specific Linux init system or service manager.
This applies to the reader, helper processes, process supervision, installation,
packaging, and required tests. Do not require systemd, `systemd-run`, `systemctl`,
or a systemd user manager. Use init-independent Linux interfaces and detect the
capabilities actually available. Fenrir must work on non-systemd distributions,
including Artix.
## Project Structure & Module Organization ## Project Structure & Module Organization
Fenrir is a Linux console screen reader with an X11 terminal mode. This checkout Fenrir is a Linux console screen reader with an X11 terminal mode. This checkout
+17
View File
@@ -147,6 +147,23 @@ a different socket or server explicitly. The sound driver does not read
PulseAudio's `client.conf`. PulseAudio's `client.conf`.
Speech output uses Speech Dispatcher's own audio configuration. Speech output uses Speech Dispatcher's own audio configuration.
Fenrir retries temporary speech initialization failures for up to 20 seconds
before acquiring keyboards. A lost Speech Dispatcher connection gets a separate
20-second recovery window in the speech worker; input remains responsive and
speech accumulated during the outage is discarded. If recovery fails, Fenrir
exits and releases its keyboards before waiting for other services to shut down.
Sound effects reconnect independently, without stopping working speech. Sound
startup calls have a three-second limit; an unresponsive attempt is abandoned
without starting overlapping attempts.
These mechanisms do not require a particular init system. Boot services running
as root keep root-owned Speech Dispatcher startup. With `sudo fenrir`, a missing
local Speech Dispatcher can instead be started as the invoking user, identified
by `SUDO_UID`; only the daemon child drops privileges. Audio servers still need
appropriate startup configuration and device/socket permissions. See
[audio recovery and verification](docs/audio-recovery.md) for limitations and
test procedures.
### X11 terminal ### X11 terminal
From an X11 terminal that supplies `WINDOWID`: From an X11 terminal that supplies `WINDOWID`:
+111
View File
@@ -0,0 +1,111 @@
# Audio startup and recovery
Fenrir handles speech and sound connections independently. The recovery code
uses Linux processes, sockets and deadlines; it does not require systemd or a
particular service manager. Explicit server, speech driver and sound driver
choices remain authoritative. Failures never select a different audio driver.
## Speech
Before opening console keyboards or an X11 terminal session, Fenrir retries
temporary speech connection failures for up to 20 seconds. Permission and
configuration errors fail without waiting through that retry window. Speech
Dispatcher's own configuration decides whether to allow daemon autostart.
When a running Speech Dispatcher connection fails, the speech worker attempts
to reconnect for up to 20 seconds. Keyboard forwarding, interruption, mute and
quit commands remain responsive. Pending speech is discarded during recovery;
it is not replayed when the connection returns. Numeric speech settings and
mute state survive recovery. Later output uses the restored connection.
If recovery fails, Fenrir reports the error and exits. Keyboard grabs are
released before waiting for UI tasks, speech or sound shutdown. Native hardware
speech drivers retain their existing failure handling; they do not start a
Speech Dispatcher daemon or select a different synthesizer.
## Root at boot and sudo sessions
A boot service running as root without `SUDO_UID` uses root's Speech Dispatcher
environment and configuration. It does not choose a logged-in user's daemon.
Root's speech and sound connections receive the same retry/recovery handling.
The audio server must be configured to start independently of graphical login
and permit the required socket and output-device access.
For `sudo fenrir`, `SUDO_UID` identifies the invoking user. If the selected local
Speech Dispatcher is missing, Fenrir looks up that account and starts only the
daemon child with that account's UID, primary group and supplementary groups.
The child receives the user's home and an owned runtime directory when present;
it does not inherit root's session bus or unrelated environment variables.
An owned `XDG_RUNTIME_DIR` is preferred for the default speech socket, followed
by the user's owned `/run/user/UID` directory, or the user's cache directory
when no runtime directory exists. Explicit speech socket/address settings
still take precedence. Fenrir retains its own console/device privileges.
Logging out may stop user-owned Speech Dispatcher or audio services. Recovery
can reconnect when the same services return; it does not keep a login session
alive or migrate to another user's audio. No session-owned daemon is killed
or restarted to force recovery.
## Sound effects
Sound effects retry failed initialization and disconnected playback in a
background worker for up to 20 seconds per outage. While unavailable, sound
cues are rejected so Fenrir can use their speech equivalents. Cues and progress
tones from an outage are discarded. The latest sound volume is applied to a
restored connection.
Each sound-driver open attempt has a three-second deadline. Some native/library
calls cannot be interrupted: if an attempt remains stuck, Fenrir stops opening
new connections for that player. At most one abandoned opener remains, and it
closes its result if it eventually returns. Working speech and keyboard input
continue. Exhausted sound recovery disables sound effects for that run.
## Limits
Retries cannot repair missing services, incorrect permissions, a wrong server,
or a missing playback device. A successful Speech Dispatcher protocol exchange
also does not prove audible output: an output module can accept a request and
fail inside its own audio backend. PipeWire, PulseAudio and Speech Dispatcher
still need appropriate configuration. In particular, PulseAudio's libpulse
autospawn behavior does not imply that a PipeWire service will start itself.
## Verification
Automated checks use private sockets and fake audio backends, without taking
over the active reader:
```sh
go test -race ./internal/audio ./internal/speech ./internal/sound ./internal/app
go test ./...
```
For optional isolated Speech Dispatcher startup checks using its dummy module:
```sh
FENRIR_NATIVE_SPEECH_STARTUP=1 go test ./internal/speech -run TestNativeSpeechDispatcherColdStart -count=1
```
Before submission, review the diff and test on an affected machine with a
recovery path, such as SSH or another accessible console. Keep only one reader
on a console. Source tests do not prove installed behavior or audible output.
1. Boot without a GUI and before user login, with Fenrir started by the normal
boot service. Verify speech, startup sounds, typing, review and quit. Repeat
with the audio server delayed briefly: speech should recover if the required
service becomes available inside the retry window.
2. Check ordinary `sudo fenrir` startup with the user's Speech Dispatcher both
already running and initially absent. Check daemon ownership and confirm
speech and sound. No root-owned daemon should be created in the user's
runtime directory. Test a console session without systemd as well.
3. With a recovery connection available, interrupt the configured speech/audio
service using that system's normal service controls. While it is unavailable,
check ordinary typing, Control, mute and Fenrir's quit shortcut. Restore the
service and check fresh review output, volume/rate/pitch and absence of old
queued announcements. A permanent speech failure must release the keyboard;
a sound-only failure must leave speech working.
4. Repeat ordinary X11 terminal startup, review, interruption and exit, plus
explicit ALSA sound output where available. Check settings changes during
recovery and verify that saved settings are not silently rewritten.
5. For user-owned audio, log out and back in with the recovery connection still
available. Verify behavior against that machine's session-lifetime policy;
this is separate from root-owned pre-login audio.
+79
View File
@@ -0,0 +1,79 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package app
import (
"context"
"errors"
"io"
"sync"
"syscall"
"testing"
"time"
"git.stormux.org/storm/fenrir/internal/command"
"git.stormux.org/storm/fenrir/internal/input"
"git.stormux.org/storm/fenrir/internal/screen"
"git.stormux.org/storm/fenrir/internal/speech"
)
type unavailableSpeech struct {
entered chan struct{}
once sync.Once
fatal bool
}
func (f *unavailableSpeech) Say(string) error { return io.EOF }
func (f *unavailableSpeech) Configure(speech.Parameters) error { return nil }
func (f *unavailableSpeech) Set_pitch(int) error { return nil }
func (f *unavailableSpeech) Cancel() error { return nil }
func (f *unavailableSpeech) Close() {}
func (f *unavailableSpeech) Reopen(ctx context.Context) (speech.Transport, error) {
f.once.Do(func() { close(f.entered) })
if f.fatal {
return nil, syscall.EACCES
}
<-ctx.Done()
return nil, ctx.Err()
}
func TestAudioRecoveryKeepsKeyboardCommandsAndReleasesOnFailure(t *testing.T) {
for _, fatal := range []bool{false, true} {
broken := &unavailableSpeech{entered: make(chan struct{}), fatal: fatal}
voice := speech.New(broken)
keys := &fakeKeys{events: make(chan error, 1)}
polls := 0
keys.poll = func(_ bool, handle func(input.Decision)) error {
polls++
if polls > 1 {
handle(input.Decision{Forward: true, Interrupt: true, Keypress: true})
handle(input.Decision{Command: command.Quit})
}
return nil
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
result := make(chan error, 1)
go func() {
result <- run(ctx, &fakeSource{snapshot: screen.Snapshot{Terminal: 1, Active: true, Rows: 1, Cols: 5, Cells: []rune("ready")}}, keys, voice, nil)
}()
select {
case <-broken.entered:
if !fatal {
keys.events <- nil
}
case <-ctx.Done():
t.Fatal("fixture did not begin speech recovery")
}
select {
case err := <-result:
if fatal && !errors.Is(err, syscall.EACCES) || !fatal && (err != nil || polls < 2) || !keys.closed {
t.Fatal("audio failure trapped input", err, polls, keys.closed)
}
case <-ctx.Done():
t.Fatal("reader could not quit or release keyboards during audio recovery")
}
cancel()
voice.Close()
}
}
+12 -1
View File
@@ -42,11 +42,20 @@ func Run(ctx context.Context, source Source, keyboards Keyboards, voices Speaker
} }
func run(ctx context.Context, source Source, keyboards Keyboards, voices Speaker, ticks <-chan time.Time, options ...RunOptions) (result error) { func run(ctx context.Context, source Source, keyboards Keyboards, voices Speaker, ticks <-chan time.Time, options ...RunOptions) (result error) {
keysClosed := false
var keyboardCloseError error
releaseKeyboards := func() {
if !keysClosed {
keysClosed = true
keyboardCloseError = keyboards.Close()
}
}
defer func() { defer func() {
if errors.Is(result, terminal.ErrClosed) { if errors.Is(result, terminal.ErrClosed) {
result = nil result = nil
} }
result = errors.Join(result, keyboards.Close()) releaseKeyboards()
result = errors.Join(result, keyboardCloseError)
}() }()
opts := RunOptions{Settings: config.Defaults()} opts := RunOptions{Settings: config.Defaults()}
if len(options) > 0 { if len(options) > 0 {
@@ -62,6 +71,8 @@ func run(ctx context.Context, source Source, keyboards Keyboards, voices Speaker
selected.adapter.Close() selected.adapter.Close()
} }
}() }()
// Release grabs before waiting for UI tasks, adapters or audio shutdown.
defer releaseKeyboards()
return a.loop(ticks) return a.loop(ticks)
} }
+5 -1
View File
@@ -83,10 +83,14 @@ func (a *application) read_screen() error {
} }
} }
diagnostic.Timing("screen processing", updated) diagnostic.Timing("screen processing", updated)
soundAvailable := a.cues != nil
if player, ok := a.cues.(interface{ Available() bool }); ok {
soundAvailable = player.Available()
}
diagnostic.Record("reader presentation", "terminal", snapshot.Terminal, "generation", snapshot.Generation, diagnostic.Record("reader presentation", "terminal", snapshot.Terminal, "generation", snapshot.Generation,
"messages", len(output), "ui_active", a.ui.active, "temporary_silence", a.temporarySilence, "messages", len(output), "ui_active", a.ui.active, "temporary_silence", a.temporarySilence,
"muted", a.muted, "speech_available", a.settings.live.Presentation.Speech.Enabled, "muted", a.muted, "speech_available", a.settings.live.Presentation.Speech.Enabled,
"progress_enabled", a.settings.live.Presentation.Progress.Enabled, "sound_enabled", a.settings.live.Presentation.Sound.Enabled, "sound_available", a.cues != nil) "progress_enabled", a.settings.live.Presentation.Progress.Enabled, "sound_enabled", a.settings.live.Presentation.Sound.Enabled, "sound_available", soundAvailable)
// VT sounds remain available while automatic reading is disabled. // VT sounds remain available while automatic reading is disabled.
if sessionChanged && snapshot.Active && len(output) == 0 { if sessionChanged && snapshot.Active && len(output) == 0 {
a.play(presentation.TTYChanged) a.play(presentation.TTYChanged)
+5 -4
View File
@@ -4,6 +4,7 @@
package app package app
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"log/slog" "log/slog"
@@ -88,19 +89,19 @@ func settings_committed(err error) bool {
return err == nil || errors.As(err, &saved) return err == nil || errors.As(err, &saved)
} }
func open_speech(selected config.Speech) (speech.Transport, config.Speech, error) { func open_speech(ctx context.Context, selected config.Speech) (speech.Transport, config.Speech, error) {
options := speech_options(selected) options := speech_options(selected)
if selected.Module == "" && selected.Language == "" && selected.Voice == "" { if selected.Module == "" && selected.Language == "" && selected.Voice == "" {
client, err := speech.Open_driver(selected.Driver, options) client, err := speech.Open_driver_context(ctx, selected.Driver, options)
return client, selected, err return client, selected, err
} }
client, err := prepare_speech(selected) client, err := speech.Prepare_context(ctx, selected.Driver, options)
if err == nil { if err == nil {
return client, selected, nil return client, selected, nil
} }
slog.Warn("saved speech selection unavailable; using server default", "error", err, "at", time.Now()) slog.Warn("saved speech selection unavailable; using server default", "error", err, "at", time.Now())
selected.Module, selected.Language, selected.Voice = "", "", "" selected.Module, selected.Language, selected.Voice = "", "", ""
options.Module, options.Language, options.Voice = "", "", "" options.Module, options.Language, options.Voice = "", "", ""
client, err = speech.Open_driver(selected.Driver, options) client, err = speech.Open_driver_context(ctx, selected.Driver, options)
return client, selected, err return client, selected, err
} }
+20 -6
View File
@@ -26,6 +26,7 @@ import (
"path/filepath" "path/filepath"
"time" "time"
"git.stormux.org/storm/fenrir/internal/audio"
"git.stormux.org/storm/fenrir/internal/clipboard" "git.stormux.org/storm/fenrir/internal/clipboard"
"git.stormux.org/storm/fenrir/internal/command" "git.stormux.org/storm/fenrir/internal/command"
"git.stormux.org/storm/fenrir/internal/config" "git.stormux.org/storm/fenrir/internal/config"
@@ -126,26 +127,39 @@ func start(ctx context.Context, settings config.Settings, terminalMode bool, con
var voices Speaker = disabledVoice{} var voices Speaker = disabledVoice{}
var effectiveSpeech *config.Speech var effectiveSpeech *config.Speech
if settings.Presentation.Speech.Enabled { if settings.Presentation.Speech.Enabled {
client, selected, err := open_speech(settings.Presentation.Speech) var client speech.Transport
var selected config.Speech
waitingForSpeech := false
retryContext, cancel := context.WithTimeout(ctx, audio.RecoveryTimeout)
err := audio.Retry(retryContext, func(ctx context.Context) error {
var err error
client, selected, err = open_speech(ctx, settings.Presentation.Speech)
if err != nil && !waitingForSpeech && audio.Temporary(err) {
waitingForSpeech = true
slog.Warn("speech unavailable; waiting before acquiring keyboards", "error", err, "at", time.Now())
}
return err
}, audio.Temporary)
cancel()
effectiveSpeech = &selected effectiveSpeech = &selected
if err != nil { if err != nil {
return err return err
} }
if waitingForSpeech {
slog.Info("speech connection ready", "at", time.Now())
}
voices = speech.New(client) voices = speech.New(client)
voices.Configure(speech_parameters(settings)) voices.Configure(speech_parameters(settings))
} }
defer voices.Close() defer voices.Close()
var cues SoundPlayer var cues SoundPlayer
var player *sound.Service var player *sound.Player
if settings.Presentation.Sound.Enabled { if settings.Presentation.Sound.Enabled {
home, err := os.UserHomeDir() home, err := os.UserHomeDir()
if err != nil { if err != nil {
return err return err
} }
player, err = sound.Open(sound.Theme_paths(home, settings.Presentation.Sound.Theme), sound.Options{Driver: settings.Presentation.Sound.Driver, Device: settings.Presentation.Sound.Device, Volume: settings.Presentation.Sound.Volume}) player = sound.Open_recovering(ctx, sound.Theme_paths(home, settings.Presentation.Sound.Theme), sound.Options{Driver: settings.Presentation.Sound.Driver, Device: settings.Presentation.Sound.Device, Volume: settings.Presentation.Sound.Volume})
if err != nil {
slog.Warn("sound unavailable or incomplete", "error", err, "at", time.Now())
}
if player != nil { if player != nil {
defer player.Close() defer player.Close()
cues = player cues = player
+58
View File
@@ -0,0 +1,58 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
// Package audio contains the shared, init-independent audio retry policy.
package audio
import (
"context"
"errors"
"io"
"net"
"os"
"os/exec"
"syscall"
"time"
)
const RecoveryTimeout = 20 * time.Second
// Temporary excludes permission, executable and protocol/configuration errors.
func Temporary(err error) bool {
if errors.Is(err, context.Canceled) {
return false
}
var executableError *exec.Error
var pathError *os.PathError
if errors.As(err, &executableError) || errors.As(err, &pathError) && pathError.Op == "fork/exec" {
return false
}
var networkError net.Error
return errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) ||
errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ECONNREFUSED) ||
errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.EPIPE) ||
errors.Is(err, context.DeadlineExceeded) || errors.As(err, &networkError) && networkError.Timeout()
}
// Retry never sleeps or performs I/O on the input loop. Callers supply a
// deadline and must make each attempt honor its context.
func Retry(ctx context.Context, attempt func(context.Context) error, retryable func(error) bool) error {
delay := 100 * time.Millisecond
for {
if err := ctx.Err(); err != nil {
return err
}
err := attempt(ctx)
if err == nil || !retryable(err) {
return err
}
timer := time.NewTimer(delay)
select {
case <-ctx.Done():
timer.Stop()
return errors.Join(ctx.Err(), err)
case <-timer.C:
}
delay = min(2*delay, time.Second)
}
}
+51
View File
@@ -0,0 +1,51 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package audio
import (
"context"
"errors"
"io"
"net"
"os/exec"
"syscall"
"testing"
"time"
)
func TestRetryRecoversAndPreservesPermanentErrors(t *testing.T) {
attempts := 0
err := Retry(context.Background(), func(context.Context) error {
attempts++
if attempts < 2 {
return &net.OpError{Op: "dial", Net: "unix", Err: syscall.ENOENT}
}
return nil
}, Temporary)
if err != nil || attempts != 2 {
t.Fatal(err, attempts)
}
for _, problem := range []error{syscall.EACCES, errors.New("invalid configuration"), &exec.Error{Name: "missing-speech-daemon", Err: exec.ErrNotFound}} {
attempts = 0
err := Retry(context.Background(), func(context.Context) error { attempts++; return problem }, Temporary)
if !errors.Is(err, problem) || attempts != 1 {
t.Fatal(err, attempts)
}
}
}
func TestRetryCancellationStopsBackoffAndDeadlinePreservesCause(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
attempts := 0
err := Retry(ctx, func(context.Context) error { attempts++; cancel(); return io.EOF }, Temporary)
if !errors.Is(err, context.Canceled) || attempts != 1 {
t.Fatal(err, attempts)
}
ctx, cancel = context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
err = Retry(ctx, func(context.Context) error { return io.EOF }, Temporary)
if !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, io.EOF) {
t.Fatal(err)
}
}
+210
View File
@@ -0,0 +1,210 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package sound
import (
"context"
"errors"
"fmt"
"log/slog"
"math"
"sync"
"sync/atomic"
"time"
"git.stormux.org/storm/fenrir/internal/audio"
"git.stormux.org/storm/fenrir/internal/presentation"
)
type soundBackend interface {
Play(presentation.Cue) bool
Play_progress(float64) bool
Play_indentation(int) bool
Discard_progress()
Clear_progress()
Set_volume(float64)
Errors() <-chan error
Finish(context.Context) error
Close()
}
type soundConnection struct{ backend soundBackend }
var errSoundAttemptStalled = errors.New("sound driver open did not return before its deadline")
// Player keeps connection work away from cue admission and keyboard handling.
// Missing output rejects cues so the caller can present their speech equivalent.
type Player struct {
connection atomic.Pointer[soundConnection]
volume atomic.Uint64
cancel context.CancelFunc
done chan struct{}
errors chan error
once sync.Once
}
func Open_recovering(ctx context.Context, roots []string, options Options) *Player {
return new_player(ctx, options.Volume, func() (soundBackend, error) {
service, err := Open(roots, options)
if service == nil {
return nil, err
}
if err != nil {
slog.Warn("sound unavailable or incomplete", "error", err, "at", time.Now())
}
return service, nil
})
}
func new_player(ctx context.Context, volume float64, open func() (soundBackend, error)) *Player {
ctx, cancel := context.WithCancel(ctx)
p := &Player{cancel: cancel, done: make(chan struct{}), errors: make(chan error, 1)}
p.volume.Store(math.Float64bits(volume))
// Even a driver whose connection or stream-start call ignores cancellation
// cannot hold startup forever. Never retry an attempt that is still stuck.
backend, err := sound_attempt(ctx, open)
if err == nil {
p.connection.Store(&soundConnection{backend})
backend.Set_volume(volume)
} else {
slog.Warn("sound unavailable or incomplete", "reconnecting", true, "error", err, "at", time.Now())
}
go p.run(ctx, open, err)
return p
}
// A timed-out native/library call is quarantined once. A late result is closed
// by its own worker; no further opener goroutines are started for this player.
func sound_attempt(ctx context.Context, open func() (soundBackend, error)) (soundBackend, error) {
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
type result struct {
backend soundBackend
err error
}
results := make(chan result)
go func() {
backend, err := open()
select {
case results <- result{backend, err}:
case <-ctx.Done():
if backend != nil {
backend.Close()
}
}
}()
select {
case result := <-results:
return result.backend, result.err
case <-ctx.Done():
return nil, errors.Join(errSoundAttemptStalled, ctx.Err())
}
}
func (p *Player) run(ctx context.Context, open func() (soundBackend, error), initialError error) {
defer close(p.done)
problem := initialError
for {
connection := p.connection.Load()
if connection == nil {
if errors.Is(problem, errSoundAttemptStalled) || errors.Is(problem, context.Canceled) {
if ctx.Err() == nil {
p.errors <- fmt.Errorf("sound initialization timed out: %w", problem)
}
return
}
retryContext, cancel := context.WithTimeout(ctx, audio.RecoveryTimeout)
var backend soundBackend
err := audio.Retry(retryContext, func(ctx context.Context) error {
var err error
backend, err = sound_attempt(ctx, open)
return err
}, func(err error) bool {
return !errors.Is(err, errSoundAttemptStalled) && !errors.Is(err, context.Canceled)
})
cancel()
if err != nil {
if ctx.Err() == nil {
p.errors <- fmt.Errorf("sound recovery failed: %w", err)
}
return
}
connection = &soundConnection{backend}
p.connection.Store(connection)
backend.Set_volume(math.Float64frombits(p.volume.Load()))
slog.Info("sound connection restored", "at", time.Now())
}
select {
case <-ctx.Done():
return
case problem = <-connection.backend.Errors():
p.connection.Store(nil)
connection.backend.Close()
slog.Warn("sound connection lost; reconnecting", "error", problem, "at", time.Now())
}
}
}
func (p *Player) Errors() <-chan error { return p.errors }
func (p *Player) Available() bool { return p.connection.Load() != nil }
func (p *Player) Play(cue presentation.Cue) bool {
if connection := p.connection.Load(); connection != nil {
return connection.backend.Play(cue)
}
return false
}
func (p *Player) Play_progress(value float64) bool {
if connection := p.connection.Load(); connection != nil {
return connection.backend.Play_progress(value)
}
return false
}
func (p *Player) Play_indentation(level int) bool {
if connection := p.connection.Load(); connection != nil {
return connection.backend.Play_indentation(level)
}
return false
}
func (p *Player) Discard_progress() {
if connection := p.connection.Load(); connection != nil {
connection.backend.Discard_progress()
}
}
func (p *Player) Clear_progress() {
if connection := p.connection.Load(); connection != nil {
connection.backend.Clear_progress()
}
}
func (p *Player) Set_volume(volume float64) {
p.volume.Store(math.Float64bits(volume))
if connection := p.connection.Load(); connection != nil {
connection.backend.Set_volume(volume)
}
}
func (p *Player) Finish(ctx context.Context) error {
p.cancel()
<-p.done
if connection := p.connection.Load(); connection != nil {
return connection.backend.Finish(ctx)
}
return nil
}
func (p *Player) Close() {
p.once.Do(func() {
p.cancel()
<-p.done
if connection := p.connection.Swap(nil); connection != nil {
connection.backend.Close()
}
})
}
+168
View File
@@ -0,0 +1,168 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package sound
import (
"context"
"errors"
"math"
"sync"
"sync/atomic"
"syscall"
"testing"
"time"
"git.stormux.org/storm/fenrir/internal/presentation"
)
type fakeBackend struct {
errors chan error
closed chan struct{}
once sync.Once
plays atomic.Int32
volume atomic.Uint64
}
func fake_backend() *fakeBackend {
return &fakeBackend{errors: make(chan error, 1), closed: make(chan struct{})}
}
func (f *fakeBackend) Play(presentation.Cue) bool { f.plays.Add(1); return true }
func (f *fakeBackend) Play_progress(float64) bool { return true }
func (f *fakeBackend) Play_indentation(int) bool { return true }
func (f *fakeBackend) Discard_progress() {}
func (f *fakeBackend) Clear_progress() {}
func (f *fakeBackend) Set_volume(volume float64) { f.volume.Store(math.Float64bits(volume)) }
func (f *fakeBackend) Errors() <-chan error { return f.errors }
func (f *fakeBackend) Finish(context.Context) error { return nil }
func (f *fakeBackend) Close() { f.once.Do(func() { close(f.closed) }) }
func wait_player(t *testing.T, player *Player, backend *fakeBackend) {
t.Helper()
deadline := time.NewTimer(time.Second)
defer deadline.Stop()
tick := time.NewTicker(time.Millisecond)
defer tick.Stop()
for {
if connection := player.connection.Load(); connection != nil && connection.backend == backend && math.Float64frombits(backend.volume.Load()) == math.Float64frombits(player.volume.Load()) {
return
}
select {
case <-deadline.C:
t.Fatal("sound connection did not recover")
case <-tick.C:
}
}
}
func TestSoundPlayerRecoversStartupAndDisconnectWithoutReplayingCues(t *testing.T) {
first, second := fake_backend(), fake_backend()
firstGate, secondGate := make(chan struct{}), make(chan struct{})
attempts := make(chan int, 8)
count := 0
player := new_player(context.Background(), 0.7, func() (soundBackend, error) {
count++
attempts <- count
switch count {
case 1:
// A completed library request timeout is safe to retry. It is
// distinct from an opener that is still running after our deadline.
return nil, context.DeadlineExceeded
case 2:
<-firstGate
return first, nil
default:
<-secondGate
return second, nil
}
})
defer player.Close()
<-attempts
<-attempts
if player.Play(presentation.Ready) || player.Play_progress(50) || player.Play_indentation(2) {
t.Fatal("offline sound suppressed speech fallback")
}
player.Set_volume(0.4)
close(firstGate)
wait_player(t, player, first)
if !player.Play(presentation.Ready) {
t.Fatal("recovered sound rejected cue")
}
first.errors <- errors.New("server disconnected")
<-attempts
if player.Play(presentation.Error) {
t.Fatal("disconnected sound accepted stale cue")
}
player.Set_volume(0.2)
close(secondGate)
wait_player(t, player, second)
if second.plays.Load() != 0 || first.plays.Load() != 1 {
t.Fatal("recovery replayed old cues", first.plays.Load(), second.plays.Load())
}
select {
case <-first.closed:
default:
t.Fatal("old sound connection leaked")
}
}
func TestUnresponsiveSoundAttemptIsCancelledAndLateResultClosed(t *testing.T) {
gate := make(chan struct{})
backend := fake_backend()
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
started := time.Now()
got, err := sound_attempt(ctx, func() (soundBackend, error) { <-gate; return backend, nil })
if got != nil || !errors.Is(err, context.DeadlineExceeded) || time.Since(started) > time.Second {
t.Fatal("sound attempt blocked caller", got, err)
}
close(gate)
select {
case <-backend.closed:
case <-time.After(time.Second):
t.Fatal("late sound result leaked")
}
attempts := 0
player := new_player(context.Background(), 0.7, func() (soundBackend, error) {
attempts++
return nil, errors.Join(errSoundAttemptStalled, context.DeadlineExceeded)
})
defer player.Close()
select {
case <-player.Errors():
case <-time.After(time.Second):
t.Fatal("unresponsive attempt was retried")
}
if attempts != 1 {
t.Fatal("overlapping sound opener calls", attempts)
}
}
func TestSoundPlayerCloseDoesNotWaitForHungReconnect(t *testing.T) {
backend := fake_backend()
gate, entered := make(chan struct{}), make(chan struct{})
attempts := 0
player := new_player(context.Background(), 0.7, func() (soundBackend, error) {
attempts++
if attempts == 1 {
return nil, syscall.ENOENT
}
close(entered)
<-gate
return backend, nil
})
<-entered
closed := make(chan struct{})
go func() { player.Close(); close(closed) }()
select {
case <-closed:
case <-time.After(time.Second):
t.Fatal("shutdown waited for hung sound server")
}
close(gate)
select {
case <-backend.closed:
case <-time.After(time.Second):
t.Fatal("late sound connection leaked")
}
}
+12 -2
View File
@@ -8,6 +8,7 @@ import (
"fmt" "fmt"
"net" "net"
"os" "os"
"os/user"
"path/filepath" "path/filepath"
"strconv" "strconv"
"strings" "strings"
@@ -22,8 +23,17 @@ type serverAddress struct {
// runtime directory selection for console sessions. // runtime directory selection for console sessions.
func default_socket_path() string { func default_socket_path() string {
if uid := os.Getenv("SUDO_UID"); uid != "" { if uid := os.Getenv("SUDO_UID"); uid != "" {
if _, err := strconv.ParseUint(uid, 10, 32); err == nil { if parsed, err := strconv.ParseUint(uid, 10, 32); err == nil {
return filepath.Join("/run/user", uid, "speech-dispatcher/speechd.sock") runtimeDirectory := os.Getenv("XDG_RUNTIME_DIR")
if !owned_directory(runtimeDirectory, uint32(parsed)) {
runtimeDirectory = filepath.Join("/run/user", uid)
}
if !owned_directory(runtimeDirectory, uint32(parsed)) {
if account, err := user.LookupId(uid); err == nil {
runtimeDirectory = filepath.Join(account.HomeDir, ".cache")
}
}
return filepath.Join(runtimeDirectory, "speech-dispatcher/speechd.sock")
} }
} }
runtimeDirectory := os.Getenv("XDG_RUNTIME_DIR") runtimeDirectory := os.Getenv("XDG_RUNTIME_DIR")
+65 -4
View File
@@ -11,6 +11,8 @@ import (
"net" "net"
"os" "os"
"os/exec" "os/exec"
"os/user"
"path/filepath"
"strconv" "strconv"
"strings" "strings"
"syscall" "syscall"
@@ -102,10 +104,6 @@ func spawn_arguments(ctx context.Context, endpoint serverAddress) ([]string, err
} }
func spawn_server(ctx context.Context, endpoint serverAddress) error { func spawn_server(ctx context.Context, endpoint serverAddress) error {
// Never create a root-owned server in the invoking user's runtime directory.
if uid := os.Getenv("SUDO_UID"); uid != "" && uid != strconv.Itoa(os.Geteuid()) {
return fmt.Errorf("start Speech Dispatcher as the invoking user before running Fenrir with sudo")
}
spawnContext, cancel := context.WithTimeout(ctx, commandTimeout) spawnContext, cancel := context.WithTimeout(ctx, commandTimeout)
defer cancel() defer cancel()
args, err := spawn_arguments(spawnContext, endpoint) args, err := spawn_arguments(spawnContext, endpoint)
@@ -117,6 +115,24 @@ func spawn_server(ctx context.Context, endpoint serverAddress) error {
command = "speech-dispatcher" command = "speech-dispatcher"
} }
process := exec.CommandContext(spawnContext, command, args...) process := exec.CommandContext(spawnContext, command, args...)
if uid := os.Getenv("SUDO_UID"); uid != "" && uid != strconv.Itoa(os.Geteuid()) {
// Drop credentials only in the child. Never run a root-owned daemon in
// another user's session or change the reader's device privileges.
if os.Geteuid() != 0 {
return fmt.Errorf("cannot start Speech Dispatcher as the invoking user without root privileges")
}
account, err := user.LookupId(uid)
if err != nil {
return fmt.Errorf("look up invoking user for Speech Dispatcher: %w", err)
}
groups, err := account.GroupIds()
if err != nil {
return fmt.Errorf("look up invoking user's groups: %w", err)
}
if err := configure_sudo_spawn(process, account, groups); err != nil {
return err
}
}
process.WaitDelay = 100 * time.Millisecond process.WaitDelay = 100 * time.Millisecond
output, err := process.CombinedOutput() output, err := process.CombinedOutput()
if err != nil { if err != nil {
@@ -124,3 +140,48 @@ func spawn_server(ctx context.Context, endpoint serverAddress) error {
} }
return nil return nil
} }
func configure_sudo_spawn(process *exec.Cmd, account *user.User, groups []string) error {
uid, uidErr := strconv.ParseUint(account.Uid, 10, 32)
gid, gidErr := strconv.ParseUint(account.Gid, 10, 32)
if uidErr != nil || gidErr != nil || uid == 0 || !filepath.IsAbs(account.HomeDir) {
return fmt.Errorf("invalid invoking user for Speech Dispatcher")
}
credentials := &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}
for _, group := range groups {
id, err := strconv.ParseUint(group, 10, 32)
if err != nil {
return fmt.Errorf("invalid invoking user's group: %w", err)
}
credentials.Groups = append(credentials.Groups, uint32(id))
}
process.SysProcAttr = &syscall.SysProcAttr{Credential: credentials}
process.Dir = account.HomeDir
// Do not pass root's HOME, XDG paths, D-Bus address, or unrelated secrets
// into a user-owned daemon. Preserve explicit audio choices and locale.
process.Env = []string{"HOME=" + account.HomeDir, "USER=" + account.Username,
"LOGNAME=" + account.Username, "PATH=/usr/local/bin:/usr/bin:/bin"}
for _, entry := range os.Environ() {
key, _, _ := strings.Cut(entry, "=")
if key == "LANG" || strings.HasPrefix(key, "LC_") || key == "PULSE_SERVER" || key == "PULSE_COOKIE" {
process.Env = append(process.Env, entry)
}
}
runtimeDir := filepath.Join("/run/user", account.Uid)
if candidate := os.Getenv("XDG_RUNTIME_DIR"); owned_directory(candidate, uint32(uid)) {
runtimeDir = candidate
}
if owned_directory(runtimeDir, uint32(uid)) {
process.Env = append(process.Env, "XDG_RUNTIME_DIR="+runtimeDir)
}
return nil
}
func owned_directory(path string, uid uint32) bool {
info, err := os.Stat(path)
if err != nil || !info.IsDir() {
return false
}
stat, ok := info.Sys().(*syscall.Stat_t)
return ok && stat.Uid == uid
}
+4 -1
View File
@@ -149,7 +149,10 @@ func TestDiscoveryDoesNotSpawn(t *testing.T) {
} }
} }
func TestSudoDoesNotSpawnAsAnotherUser(t *testing.T) { func TestUnprivilegedProcessDoesNotImpersonateSudoUser(t *testing.T) {
if os.Geteuid() == 0 {
t.Skip("requires an unprivileged caller")
}
t.Setenv("SUDO_UID", strconv.Itoa(os.Geteuid()+1)) t.Setenv("SUDO_UID", strconv.Itoa(os.Geteuid()+1))
err := spawn_server(context.Background(), serverAddress{"unix", "/tmp/custom.sock"}) err := spawn_server(context.Background(), serverAddress{"unix", "/tmp/custom.sock"})
if err == nil || !strings.Contains(err.Error(), "invoking user") { if err == nil || !strings.Contains(err.Error(), "invoking user") {
+11 -4
View File
@@ -4,6 +4,7 @@
package speech package speech
import ( import (
"context"
"fmt" "fmt"
"os" "os"
"sort" "sort"
@@ -45,7 +46,7 @@ func Discover(options Options) (Catalog, error) {
if err != nil { if err != nil {
return result, err return result, err
} }
defer w.conn.Close() defer w.close()
deadline := time.Now().Add(4 * time.Second) deadline := time.Now().Add(4 * time.Second)
query := func(command string) ([]string, error) { query := func(command string) ([]string, error) {
if time.Now().After(deadline) { if time.Now().After(deadline) {
@@ -176,6 +177,12 @@ func Language_voices(voices []Voice, language string) []Voice {
// Prepare checks membership because SSIP SET may accept unknown synthesis // Prepare checks membership because SSIP SET may accept unknown synthesis
// names without verifying them. The resulting client has not spoken anything. // names without verifying them. The resulting client has not spoken anything.
func Prepare(driver string, options Options) (Transport, error) { func Prepare(driver string, options Options) (Transport, error) {
return Prepare_context(context.Background(), driver, options)
}
// Prepare_context bounds discovery and connection setup together.
func Prepare_context(ctx context.Context, driver string, options Options) (Transport, error) {
options.context = ctx
if err := Validate_selection(options.Module, options.Language, options.Voice); err != nil { if err := Validate_selection(options.Module, options.Language, options.Voice); err != nil {
return nil, err return nil, err
} }
@@ -185,7 +192,7 @@ func Prepare(driver string, options Options) (Transport, error) {
} }
if d.Voices == nil { if d.Voices == nil {
if options.Module == "" && options.Language == "" && options.Voice == "" { if options.Module == "" && options.Language == "" && options.Voice == "" {
return Open_driver(driver, options) return Open_driver_context(ctx, driver, options)
} }
return nil, fmt.Errorf("speech driver does not support voice discovery") return nil, fmt.Errorf("speech driver does not support voice discovery")
} }
@@ -219,7 +226,7 @@ func Prepare(driver string, options Options) (Transport, error) {
if !languageFound || !voiceFound { if !languageFound || !voiceFound {
return nil, fmt.Errorf("speech language or voice is unavailable for module %q", options.Module) return nil, fmt.Errorf("speech language or voice is unavailable for module %q", options.Module)
} }
return Open_driver(driver, options) return Open_driver_context(ctx, driver, options)
} }
// A live change queries only its target, so a slow unrelated module cannot // A live change queries only its target, so a slow unrelated module cannot
@@ -232,7 +239,7 @@ func discover_voices(options Options) ([]Voice, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer w.conn.Close() defer w.close()
if options.Module != "" { if options.Module != "" {
if _, err = w.command("SET SELF OUTPUT_MODULE " + options.Module); err != nil { if _, err = w.command("SET SELF OUTPUT_MODULE " + options.Module); err != nil {
return nil, err return nil, err
+56 -13
View File
@@ -43,8 +43,20 @@ type Rejection struct{ Code string }
func (r *Rejection) Error() string { return "Speech Dispatcher rejected a command: " + r.Code } func (r *Rejection) Error() string { return "Speech Dispatcher rejected a command: " + r.Code }
type wire struct { type wire struct {
conn net.Conn conn net.Conn
reader *bufio.Reader reader *bufio.Reader
stopContext func() bool
}
func (w *wire) detach_context() {
if w.stopContext != nil {
w.stopContext()
}
}
func (w *wire) close() {
w.detach_context()
_ = w.conn.Close()
} }
func (w *wire) reply() ([]string, error) { func (w *wire) reply() ([]string, error) {
@@ -102,11 +114,14 @@ func connect_options(options Options) (*wire, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
return connect_endpoint(endpoint, options.autospawn, automatic_server_address(options.Socket)) return connect_endpoint(options.context, endpoint, options.autospawn, automatic_server_address(options.Socket))
} }
func connect_endpoint(endpoint serverAddress, autospawn, systemFallback bool) (*wire, error) { func connect_endpoint(parent context.Context, endpoint serverAddress, autospawn, systemFallback bool) (*wire, error) {
ctx, cancel := context.WithTimeout(context.Background(), 3*commandTimeout) if parent == nil {
parent = context.Background()
}
ctx, cancel := context.WithTimeout(parent, 3*commandTimeout)
defer cancel() defer cancel()
dialer := &net.Dialer{Timeout: commandTimeout} dialer := &net.Dialer{Timeout: commandTimeout}
dial := serverDialer(dialer.DialContext) dial := serverDialer(dialer.DialContext)
@@ -123,7 +138,7 @@ func connect_endpoint(endpoint serverAddress, autospawn, systemFallback bool) (*
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &wire{conn: conn, reader: bufio.NewReaderSize(conn, 4096)}, nil return &wire{conn: conn, reader: bufio.NewReaderSize(conn, 4096), stopContext: context.AfterFunc(parent, func() { _ = conn.Close() })}, nil
} }
type Client struct { type Client struct {
@@ -132,11 +147,13 @@ type Client struct {
sendMu sync.Mutex sendMu sync.Mutex
controlMu sync.Mutex controlMu sync.Mutex
closeOnce sync.Once closeOnce sync.Once
options Options
} }
type Options struct { type Options struct {
// Only Prepare enables startup during its prerequisite selection queries. // Only Prepare enables startup during its prerequisite selection queries.
autospawn bool autospawn bool
context context.Context
Socket string Socket string
Rate int Rate int
Pitch *int Pitch *int
@@ -174,15 +191,15 @@ func Open(options Options) (*Client, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
data, err := connect_endpoint(endpoint, true, automatic_server_address(options.Socket)) data, err := connect_endpoint(options.context, endpoint, true, automatic_server_address(options.Socket))
if err != nil { if err != nil {
return nil, fmt.Errorf("connect Speech Dispatcher at %s: %w", endpoint.address, err) return nil, fmt.Errorf("connect Speech Dispatcher at %s: %w", endpoint.address, err)
} }
client := &Client{data: data} client := &Client{data: data, options: options}
fail := func(err error) (*Client, error) { fail := func(err error) (*Client, error) {
data.conn.Close() data.close()
if client.control != nil { if client.control != nil {
client.control.conn.Close() client.control.close()
} }
return nil, err return nil, err
} }
@@ -229,13 +246,31 @@ func Open(options Options) (*Client, error) {
if endpoint.network == "unix" { if endpoint.network == "unix" {
endpoint.address = data.conn.RemoteAddr().String() endpoint.address = data.conn.RemoteAddr().String()
} }
client.control, err = connect_endpoint(endpoint, false, false) client.control, err = connect_endpoint(options.context, endpoint, false, false)
if err != nil { if err != nil {
return fail(err) return fail(err)
} }
if _, err := client.control.command(`SET SELF CLIENT_NAME "fenrir:fenrir:control"`); err != nil { if _, err := client.control.command(`SET SELF CLIENT_NAME "fenrir:fenrir:control"`); err != nil {
return fail(err) return fail(err)
} }
data.detach_context()
client.control.detach_context()
if options.context != nil && options.context.Err() != nil {
client.abort()
return nil, options.context.Err()
}
client.options.context = nil
return client, nil
}
// Reopen reconnects only the selected server and synthesis options.
func (c *Client) Reopen(ctx context.Context) (Transport, error) {
options := c.options
options.context = ctx
client, err := Open(options)
if err != nil {
return nil, err
}
return client, nil return client, nil
} }
@@ -315,8 +350,16 @@ func (c *Client) Cancel() error {
func (c *Client) Close() { func (c *Client) Close() {
c.closeOnce.Do(func() { c.closeOnce.Do(func() {
// Closing the data socket releases a blocked sender before cancellation. // Closing the data socket releases a blocked sender before cancellation.
c.data.conn.Close() c.data.close()
c.Cancel() c.Cancel()
c.control.conn.Close() c.control.close()
}) })
} }
// abort releases blocked initialization commands without another SSIP request.
func (c *Client) abort() {
_ = c.data.conn.Close()
if c.control != nil {
_ = c.control.conn.Close()
}
}
+6 -3
View File
@@ -21,7 +21,9 @@ import (
"bufio" "bufio"
"fmt" "fmt"
"net" "net"
"os"
"path/filepath" "path/filepath"
"strconv"
"strings" "strings"
"sync" "sync"
"testing" "testing"
@@ -299,9 +301,10 @@ func TestLiveParametersUseSpeakingClientAndRejectInvalidValues(t *testing.T) {
func TestDefaultSocketForSudo(t *testing.T) { func TestDefaultSocketForSudo(t *testing.T) {
t.Setenv("SPEECHD_ADDRESS", "") t.Setenv("SPEECHD_ADDRESS", "")
t.Setenv("SPEECHD_SOCKET", "") t.Setenv("SPEECHD_SOCKET", "")
t.Setenv("SUDO_UID", "1000") runtimeDirectory := t.TempDir()
t.Setenv("XDG_RUNTIME_DIR", "/run/user/0") t.Setenv("SUDO_UID", strconv.Itoa(os.Getuid()))
if got, err := resolve_server_address(""); err != nil || got.address != "/run/user/1000/speech-dispatcher/speechd.sock" { t.Setenv("XDG_RUNTIME_DIR", runtimeDirectory)
if got, err := resolve_server_address(""); err != nil || got.address != filepath.Join(runtimeDirectory, "speech-dispatcher/speechd.sock") {
t.Fatal(got, err) t.Fatal(got, err)
} }
t.Setenv("SPEECHD_ADDRESS", "unix_socket:/tmp/example-speechd.sock") t.Setenv("SPEECHD_ADDRESS", "unix_socket:/tmp/example-speechd.sock")
+16 -3
View File
@@ -17,7 +17,10 @@
package speech package speech
import "fmt" import (
"context"
"fmt"
)
// Driver binds a stable file/CLI ID to its implementation. Availability must // Driver binds a stable file/CLI ID to its implementation. Availability must
// be a cheap dependency/hardware check, never connect, speak or start a service. // be a cheap dependency/hardware check, never connect, speak or start a service.
@@ -83,6 +86,12 @@ func Find_driver(id string) (Driver, error) {
} }
func Open_driver(id string, options Options) (Transport, error) { func Open_driver(id string, options Options) (Transport, error) {
return Open_driver_context(context.Background(), id, options)
}
// Open_driver_context bounds connection setup without binding the live client to ctx.
func Open_driver_context(ctx context.Context, id string, options Options) (Transport, error) {
options.context = ctx
driver, err := Find_driver(id) driver, err := Find_driver(id)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -93,7 +102,11 @@ func Open_driver(id string, options Options) (Transport, error) {
if err = driver.Availability(); err != nil { if err = driver.Availability(); err != nil {
return nil, fmt.Errorf("speech driver %s unavailable: %w", id, err) return nil, fmt.Errorf("speech driver %s unavailable: %w", id, err)
} }
return driver.Open(options) client, err := driver.Open(options)
if err != nil {
return nil, err
}
return client, nil
} }
// discover_modules uses a separate, bounded connection. It never changes the // discover_modules uses a separate, bounded connection. It never changes the
@@ -103,7 +116,7 @@ func discover_modules(options Options) ([]string, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer wire.conn.Close() defer wire.close()
modules, err := wire.command("LIST OUTPUT_MODULES") modules, err := wire.command("LIST OUTPUT_MODULES")
if err != nil { if err != nil {
return nil, err return nil, err
+288
View File
@@ -0,0 +1,288 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package speech
import (
"context"
"errors"
"io"
"net"
"os"
"os/exec"
"os/user"
"path/filepath"
"reflect"
"strconv"
"strings"
"sync/atomic"
"syscall"
"testing"
"time"
"git.stormux.org/storm/fenrir/internal/audio"
)
type disconnectedTransport struct {
*recordingTransport
reopen func(context.Context) (Transport, error)
closed atomic.Bool
}
func (f *disconnectedTransport) Say(string) error { return io.EOF }
func (f *disconnectedTransport) Cancel() error { return io.EOF }
func (f *disconnectedTransport) Close() { f.closed.Store(true) }
func (f *disconnectedTransport) Reopen(ctx context.Context) (Transport, error) {
return f.reopen(ctx)
}
func TestRecoveryKeepsAdmissionResponsiveAndDropsOutageSpeech(t *testing.T) {
entered, release := make(chan struct{}), make(chan struct{})
fresh := new_recording_transport()
broken := &disconnectedTransport{recordingTransport: new_recording_transport(), reopen: func(ctx context.Context) (Transport, error) {
close(entered)
select {
case <-release:
return fresh, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}}
service := New(broken)
defer service.Close()
service.Configure(Parameters{Rate: 40, Pitch: 20, Volume: 80})
service.Say("lost")
<-entered
admitted := make(chan struct{})
go func() {
service.Interrupt()
service.Configure(Parameters{Rate: 70, Pitch: 10, Volume: 90})
service.Say("discarded during outage")
service.Toggle_mute()
service.Say("discarded while muted")
close(admitted)
}()
select {
case <-admitted:
case <-time.After(time.Second):
t.Fatal("audio recovery blocked input admission")
}
close(release)
wait_event(t, fresh.changed, "configure:70:10:90")
service.Toggle_mute()
service.Say_with_pitch("fresh", 60)
wait_event(t, fresh.changed, "sent:fresh")
fresh.mu.Lock()
joined := strings.Join(fresh.events, "|")
fresh.mu.Unlock()
if strings.Contains(joined, "lost") || strings.Contains(joined, "discarded") || !strings.Contains(joined, "pitch:60|say:fresh|sent:fresh|pitch:10") || !broken.closed.Load() {
t.Fatal(joined, broken.closed.Load())
}
select {
case err := <-service.Errors():
t.Fatal("successful recovery was fatal", err)
default:
}
}
func TestRecoveryFailureIsFatalAndShutdownCancelsReopen(t *testing.T) {
for _, failure := range []bool{true, false} {
t.Run(strconv.FormatBool(failure), func(t *testing.T) {
entered := make(chan struct{})
broken := &disconnectedTransport{recordingTransport: new_recording_transport(), reopen: func(ctx context.Context) (Transport, error) {
close(entered)
if failure {
return nil, syscall.EACCES
}
<-ctx.Done()
return nil, ctx.Err()
}}
service := New(broken)
service.Say("trigger recovery")
<-entered
if failure {
select {
case err := <-service.Errors():
if !errors.Is(err, syscall.EACCES) {
t.Fatal(err)
}
case <-time.After(time.Second):
t.Fatal("permanent recovery failure was hidden")
}
}
closed := make(chan struct{})
go func() { service.Close(); close(closed) }()
select {
case <-closed:
case <-time.After(time.Second):
t.Fatal("shutdown waited for unavailable speech")
}
})
}
}
func TestSpeechRecoveryDeadlineReportsFailure(t *testing.T) {
broken := &disconnectedTransport{recordingTransport: new_recording_transport(), reopen: func(ctx context.Context) (Transport, error) {
<-ctx.Done()
return nil, ctx.Err()
}}
service := New(broken)
defer service.Close()
ctx, cancel := context.WithTimeout(service.recoveryContext, 40*time.Millisecond)
defer cancel()
service.recoveryContext = ctx
service.Say("trigger unavailable daemon")
select {
case err := <-service.Errors():
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatal(err)
}
case <-time.After(time.Second):
t.Fatal("exhausted recovery left the reader silently running")
}
}
func TestOpenContextCancelsSilentHandshakeAndDoesNotOwnLiveClient(t *testing.T) {
listener, err := net.Listen("unix", filepath.Join(t.TempDir(), "silent.sock"))
if err != nil {
t.Fatal(err)
}
defer listener.Close()
accepted := make(chan net.Conn, 1)
go func() { conn, _ := listener.Accept(); accepted <- conn }()
ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
defer cancel()
started := time.Now()
client, err := Open_driver_context(ctx, "speech-dispatcher", Options{Socket: listener.Addr().String()})
if client != nil || err == nil || time.Since(started) > time.Second {
t.Fatal("silent handshake ignored cancellation", client, err)
}
(<-accepted).Close()
server := new_fake_server(t, false)
ctx, cancel = context.WithCancel(context.Background())
client, err = Open_driver_context(ctx, "speech-dispatcher", Options{Socket: server.path})
if err != nil {
t.Fatal(err)
}
defer client.Close()
cancel()
if err := client.Say("live client survives startup context"); err != nil {
t.Fatal(err)
}
}
func TestSudoSpawnDropsChildCredentialsAndFiltersRootEnvironment(t *testing.T) {
t.Setenv("HOME", "/root")
t.Setenv("XDG_RUNTIME_DIR", "/root/runtime")
t.Setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/root/session-bus")
t.Setenv("FENRIR_TEST_SECRET", "must-not-reach-child")
t.Setenv("PULSE_SERVER", "unix:/tmp/selected-audio.sock")
command := exec.Command("speech-dispatcher", "--spawn")
account := &user.User{Uid: "2000", Gid: "2001", Username: "Username", HomeDir: "/home/Username"}
if err := configure_sudo_spawn(command, account, []string{"2001", "2002"}); err != nil {
t.Fatal(err)
}
want := &syscall.Credential{Uid: 2000, Gid: 2001, Groups: []uint32{2001, 2002}}
if !reflect.DeepEqual(command.SysProcAttr.Credential, want) || command.Dir != account.HomeDir {
t.Fatal(command.SysProcAttr, command.Dir)
}
environment := strings.Join(command.Env, "\n")
if strings.Contains(environment, "/root") || strings.Contains(environment, "FENRIR_TEST_SECRET") || !strings.Contains(environment, "HOME=/home/Username") || !strings.Contains(environment, "PULSE_SERVER=unix:/tmp/selected-audio.sock") {
t.Fatal(environment)
}
account.Uid = "0"
if err := configure_sudo_spawn(command, account, nil); err == nil {
t.Fatal("invalid invoking user selected root")
}
}
func TestStartupWithoutSudoRetriesUntilPrivateDaemonAppears(t *testing.T) {
t.Setenv("SUDO_UID", "")
t.Setenv("SPEECHD_CMD", "/nonexistent-fixture-speech-dispatcher")
socket := filepath.Join(t.TempDir(), "speech.sock")
t.Setenv("SPEECHD_ADDRESS", "unix_socket:"+socket)
t.Setenv("SPEECHD_SOCKET", "")
uid := os.Geteuid()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
failed := make(chan struct{}, 1)
result := make(chan error, 1)
var client Transport
go func() {
result <- audio.Retry(ctx, func(ctx context.Context) error {
var err error
client, err = Open_driver_context(ctx, "speech-dispatcher", Options{})
if err != nil {
select {
case failed <- struct{}{}:
default:
}
}
return err
}, audio.Temporary)
}()
select {
case <-failed:
case <-ctx.Done():
t.Fatal("initial absent daemon was not detected")
}
listener, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
}
server := new_fake_server_listener(t, false, listener)
if err := <-result; err != nil {
t.Fatal(err)
}
defer client.Close()
cancel()
if os.Geteuid() != uid {
t.Fatal("startup changed reader credentials")
}
if err := client.Say("late daemon ready"); err != nil {
t.Fatal(err)
}
wait_event(t, server.events, "say:late daemon ready")
}
func TestClientReconnectsAfterPrivateServerDropsBothConnections(t *testing.T) {
server := new_fake_server(t, false)
client, err := Open(Options{Socket: server.path})
if err != nil {
t.Fatal(err)
}
service := New(client)
defer service.Close()
service.Configure(Parameters{Rate: 30, Pitch: 10, Volume: 70})
service.Say("before disconnect")
wait_event(t, server.events, "say:before disconnect")
server.mu.Lock()
for _, conn := range server.connections {
conn.Close()
}
server.mu.Unlock()
service.Say("failed request")
deadline := time.NewTimer(time.Second)
defer deadline.Stop()
tick := time.NewTicker(time.Millisecond)
defer tick.Stop()
for {
service.mu.Lock()
service.transportMu.RLock()
restored := service.transport != nil && service.transport != client && !service.recovering
service.transportMu.RUnlock()
service.mu.Unlock()
if restored {
break
}
select {
case err := <-service.Errors():
t.Fatal("reconnection failed", err)
case <-deadline.C:
t.Fatal("both Speech Dispatcher sockets were not restored")
case <-tick.C:
}
}
service.Say("after reconnect")
wait_event(t, server.events, "say:after reconnect")
}
+111 -18
View File
@@ -18,7 +18,11 @@
package speech package speech
import ( import (
"context"
"errors" "errors"
"log/slog"
"git.stormux.org/storm/fenrir/internal/audio"
"strings" "strings"
"sync" "sync"
"time" "time"
@@ -39,6 +43,11 @@ type Transport interface {
type Service struct { type Service struct {
transport Transport transport Transport
transportMu sync.RWMutex transportMu sync.RWMutex
transportEpoch uint64
faults chan transportFailure
recoveryContext context.Context
stopRecovery context.CancelFunc
recovering bool
replacements []replacement replacements []replacement
mu sync.Mutex mu sync.Mutex
dispatch sync.Mutex dispatch sync.Mutex
@@ -52,6 +61,11 @@ type Service struct {
workers sync.WaitGroup workers sync.WaitGroup
} }
type transportFailure struct {
epoch uint64
err error
}
type replacement struct { type replacement struct {
transport Transport transport Transport
done chan error done chan error
@@ -82,7 +96,8 @@ type utterance struct {
} }
func New(transport Transport) *Service { func New(transport Transport) *Service {
s := &Service{transport: transport, wake: make(chan struct{}, 1), cancelWake: make(chan struct{}, 1), done: make(chan struct{}), errors: make(chan error, 1)} ctx, cancel := context.WithCancel(context.Background())
s := &Service{recoveryContext: ctx, stopRecovery: cancel, faults: make(chan transportFailure, 1), transport: transport, wake: make(chan struct{}, 1), cancelWake: make(chan struct{}, 1), done: make(chan struct{}), errors: make(chan error, 1)}
s.workers.Add(2) s.workers.Add(2)
go s.deliver() go s.deliver()
go s.cancel() go s.cancel()
@@ -143,7 +158,7 @@ func (s *Service) enqueue(item utterance) {
} }
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
if s.closed || s.muted { if s.closed || s.muted || s.recovering {
return return
} }
item.queuedAt = time.Now() item.queuedAt = time.Now()
@@ -189,6 +204,72 @@ func (s *Service) fail(err error) {
} }
} }
// Recovery belongs to the delivery worker. Admission and cancellation never
// wait for a new daemon, and speech from the outage is never replayed.
func (s *Service) recover_transport(problem error, parameters *Parameters) bool {
reopener, ok := s.transport.(interface {
Reopen(context.Context) (Transport, error)
})
if !ok || !audio.Temporary(problem) {
s.fail(problem)
return false
}
s.dispatch.Lock()
defer s.dispatch.Unlock()
s.mu.Lock()
s.recovering = true
s.queue = nil
s.mu.Unlock()
s.transportMu.Lock()
old := s.transport
s.transport = nil
s.transportEpoch++
s.transportMu.Unlock()
old.Close()
slog.Warn("speech connection lost; reconnecting", "error", problem, "at", time.Now())
ctx, cancel := context.WithTimeout(s.recoveryContext, audio.RecoveryTimeout)
defer cancel()
var candidate Transport
err := audio.Retry(ctx, func(ctx context.Context) error {
var err error
candidate, err = reopener.Reopen(ctx)
if err == nil && parameters != nil {
if client, ok := candidate.(*Client); ok {
stop := context.AfterFunc(ctx, client.abort)
err = candidate.Configure(*parameters)
stop()
} else {
err = candidate.Configure(*parameters)
}
}
if err != nil && candidate != nil {
candidate.Close()
candidate = nil
}
return err
}, audio.Temporary)
err = errors.Join(err, ctx.Err())
s.mu.Lock()
if err != nil || s.closed || ctx.Err() != nil {
closed := s.closed
s.mu.Unlock()
if candidate != nil {
candidate.Close()
}
if err != nil && !closed {
s.fail(errors.Join(errors.New("speech recovery failed"), err))
}
return false
}
s.transportMu.Lock()
s.transport = candidate
s.transportMu.Unlock()
s.recovering = false
s.mu.Unlock()
slog.Info("speech connection restored", "at", time.Now())
return true
}
func (s *Service) deliver() { func (s *Service) deliver() {
defer s.workers.Done() defer s.workers.Done()
normalPitch := 0 normalPitch := 0
@@ -198,6 +279,10 @@ func (s *Service) deliver() {
case <-s.done: case <-s.done:
return return
case <-s.wake: case <-s.wake:
case fault := <-s.faults:
if fault.epoch == s.transportEpoch && !s.recover_transport(fault.err, currentParameters) {
return
}
} }
for { for {
s.mu.Lock() s.mu.Lock()
@@ -211,8 +296,9 @@ func (s *Service) deliver() {
err := s.transport.Configure(parameters) err := s.transport.Configure(parameters)
s.dispatch.Unlock() s.dispatch.Unlock()
if err != nil { if err != nil {
s.fail(err) if !s.recover_transport(err, currentParameters) {
return return
}
} }
continue continue
} }
@@ -233,6 +319,7 @@ func (s *Service) deliver() {
s.transportMu.Lock() s.transportMu.Lock()
old := s.transport old := s.transport
s.transport = request.transport s.transport = request.transport
s.transportEpoch++
s.transportMu.Unlock() s.transportMu.Unlock()
s.mu.Unlock() s.mu.Unlock()
old.Close() old.Close()
@@ -262,8 +349,9 @@ func (s *Service) deliver() {
diagnostic.Timing("speech send", started, "generation", generation) diagnostic.Timing("speech send", started, "generation", generation)
s.dispatch.Unlock() s.dispatch.Unlock()
if err != nil { if err != nil {
s.fail(err) if !s.recover_transport(err, currentParameters) {
return return
}
} }
} }
} }
@@ -293,7 +381,17 @@ func (s *Service) say(item utterance, normalPitch int, normalParameters *Paramet
func (s *Service) cancel_transport() error { func (s *Service) cancel_transport() error {
s.transportMu.RLock() s.transportMu.RLock()
defer s.transportMu.RUnlock() defer s.transportMu.RUnlock()
return s.transport.Cancel() if s.transport == nil {
return nil
}
err := s.transport.Cancel()
if err != nil {
select {
case s.faults <- transportFailure{epoch: s.transportEpoch, err: err}:
default:
}
}
return err
} }
func (s *Service) cancel() { func (s *Service) cancel() {
@@ -320,21 +418,13 @@ func (s *Service) cancel() {
if !s.dispatch.TryLock() { if !s.dispatch.TryLock() {
// Stop audible speech even while a stale send is blocked. A // Stop audible speech even while a stale send is blocked. A
// second cancel after that send finishes closes the late-send race. // second cancel after that send finishes closes the late-send race.
if err := s.cancel_transport(); err != nil { _ = s.cancel_transport()
s.fail(err)
return
}
s.dispatch.Lock() s.dispatch.Lock()
} }
// With dispatch held, no send can race this cancel. The usual // With dispatch held, no send can race this cancel. The usual
// idle-send path needs only this one acknowledgement. // idle-send path needs only this one acknowledgement.
err := s.cancel_transport() _ = s.cancel_transport()
diagnostic.Timing("speech cancel barrier", started, "generation", generation) diagnostic.Timing("speech cancel barrier", started, "generation", generation)
if err != nil {
s.dispatch.Unlock()
s.fail(err)
return
}
s.mu.Lock() s.mu.Lock()
// Every intervening interrupt is covered: fresh sends have been // Every intervening interrupt is covered: fresh sends have been
// gated throughout. Coalesce them instead of canceling again. // gated throughout. Coalesce them instead of canceling again.
@@ -354,6 +444,7 @@ func (s *Service) Close() {
return return
} }
s.closed = true s.closed = true
s.stopRecovery()
s.queue = nil s.queue = nil
close(s.done) close(s.done)
pending := s.replacements pending := s.replacements
@@ -364,7 +455,9 @@ func (s *Service) Close() {
request.done <- errors.New("speech service closed") request.done <- errors.New("speech service closed")
} }
s.transportMu.RLock() s.transportMu.RLock()
s.transport.Close() if s.transport != nil {
s.transport.Close()
}
s.transportMu.RUnlock() s.transportMu.RUnlock()
s.workers.Wait() s.workers.Wait()
} }