From dc945ff39207ba968afadeeace526f825873b483 Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Tue, 3 Nov 2026 13:21:50 -0500 Subject: [PATCH] Attempt to make audio and speech discovery more robust. --- AGENTS.md | 9 + README.md | 17 ++ docs/audio-recovery.md | 111 +++++++++++ internal/app/audio_recovery_test.go | 79 ++++++++ internal/app/event_loop.go | 13 +- internal/app/screen_updates.go | 6 +- internal/app/speech_selection.go | 9 +- internal/app/startup.go | 26 ++- internal/audio/retry.go | 58 ++++++ internal/audio/retry_test.go | 51 +++++ internal/sound/player.go | 210 ++++++++++++++++++++ internal/sound/player_test.go | 168 ++++++++++++++++ internal/speech/address.go | 14 +- internal/speech/autospawn.go | 69 ++++++- internal/speech/autospawn_test.go | 5 +- internal/speech/catalog.go | 15 +- internal/speech/client.go | 69 +++++-- internal/speech/client_test.go | 9 +- internal/speech/drivers.go | 19 +- internal/speech/recovery_test.go | 288 ++++++++++++++++++++++++++++ internal/speech/service.go | 129 +++++++++++-- 21 files changed, 1314 insertions(+), 60 deletions(-) create mode 100644 docs/audio-recovery.md create mode 100644 internal/app/audio_recovery_test.go create mode 100644 internal/audio/retry.go create mode 100644 internal/audio/retry_test.go create mode 100644 internal/sound/player.go create mode 100644 internal/sound/player_test.go create mode 100644 internal/speech/recovery_test.go diff --git a/AGENTS.md b/AGENTS.md index 09d9c513..194980ec 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,14 @@ # 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 Fenrir is a Linux console screen reader with an X11 terminal mode. This checkout diff --git a/README.md b/README.md index f5da156e..72477495 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,23 @@ a different socket or server explicitly. The sound driver does not read PulseAudio's `client.conf`. 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 From an X11 terminal that supplies `WINDOWID`: diff --git a/docs/audio-recovery.md b/docs/audio-recovery.md new file mode 100644 index 00000000..cb8aeede --- /dev/null +++ b/docs/audio-recovery.md @@ -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. diff --git a/internal/app/audio_recovery_test.go b/internal/app/audio_recovery_test.go new file mode 100644 index 00000000..376d3b6b --- /dev/null +++ b/internal/app/audio_recovery_test.go @@ -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() + } +} diff --git a/internal/app/event_loop.go b/internal/app/event_loop.go index 3ad2d700..0954f25c 100644 --- a/internal/app/event_loop.go +++ b/internal/app/event_loop.go @@ -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) { + keysClosed := false + var keyboardCloseError error + releaseKeyboards := func() { + if !keysClosed { + keysClosed = true + keyboardCloseError = keyboards.Close() + } + } defer func() { if errors.Is(result, terminal.ErrClosed) { result = nil } - result = errors.Join(result, keyboards.Close()) + releaseKeyboards() + result = errors.Join(result, keyboardCloseError) }() opts := RunOptions{Settings: config.Defaults()} if len(options) > 0 { @@ -62,6 +71,8 @@ func run(ctx context.Context, source Source, keyboards Keyboards, voices Speaker selected.adapter.Close() } }() + // Release grabs before waiting for UI tasks, adapters or audio shutdown. + defer releaseKeyboards() return a.loop(ticks) } diff --git a/internal/app/screen_updates.go b/internal/app/screen_updates.go index 8bcbb43a..f882463f 100644 --- a/internal/app/screen_updates.go +++ b/internal/app/screen_updates.go @@ -83,10 +83,14 @@ func (a *application) read_screen() error { } } 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, "messages", len(output), "ui_active", a.ui.active, "temporary_silence", a.temporarySilence, "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. if sessionChanged && snapshot.Active && len(output) == 0 { a.play(presentation.TTYChanged) diff --git a/internal/app/speech_selection.go b/internal/app/speech_selection.go index 17f5f3f0..61128885 100644 --- a/internal/app/speech_selection.go +++ b/internal/app/speech_selection.go @@ -4,6 +4,7 @@ package app import ( + "context" "errors" "fmt" "log/slog" @@ -88,19 +89,19 @@ func settings_committed(err error) bool { 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) 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 } - client, err := prepare_speech(selected) + client, err := speech.Prepare_context(ctx, selected.Driver, options) if err == nil { return client, selected, nil } slog.Warn("saved speech selection unavailable; using server default", "error", err, "at", time.Now()) selected.Module, selected.Language, selected.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 } diff --git a/internal/app/startup.go b/internal/app/startup.go index 7a42bf4f..0c848d1e 100644 --- a/internal/app/startup.go +++ b/internal/app/startup.go @@ -26,6 +26,7 @@ import ( "path/filepath" "time" + "git.stormux.org/storm/fenrir/internal/audio" "git.stormux.org/storm/fenrir/internal/clipboard" "git.stormux.org/storm/fenrir/internal/command" "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 effectiveSpeech *config.Speech 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 if err != nil { return err } + if waitingForSpeech { + slog.Info("speech connection ready", "at", time.Now()) + } voices = speech.New(client) voices.Configure(speech_parameters(settings)) } defer voices.Close() var cues SoundPlayer - var player *sound.Service + var player *sound.Player if settings.Presentation.Sound.Enabled { home, err := os.UserHomeDir() if err != nil { 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}) - if err != nil { - slog.Warn("sound unavailable or incomplete", "error", err, "at", time.Now()) - } + 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 player != nil { defer player.Close() cues = player diff --git a/internal/audio/retry.go b/internal/audio/retry.go new file mode 100644 index 00000000..9c7b6040 --- /dev/null +++ b/internal/audio/retry.go @@ -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) + } +} diff --git a/internal/audio/retry_test.go b/internal/audio/retry_test.go new file mode 100644 index 00000000..e827fa4b --- /dev/null +++ b/internal/audio/retry_test.go @@ -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) + } +} diff --git a/internal/sound/player.go b/internal/sound/player.go new file mode 100644 index 00000000..8b1fa9fb --- /dev/null +++ b/internal/sound/player.go @@ -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() + } + }) +} diff --git a/internal/sound/player_test.go b/internal/sound/player_test.go new file mode 100644 index 00000000..5f498573 --- /dev/null +++ b/internal/sound/player_test.go @@ -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") + } +} diff --git a/internal/speech/address.go b/internal/speech/address.go index a1d6a687..3be908c0 100644 --- a/internal/speech/address.go +++ b/internal/speech/address.go @@ -8,6 +8,7 @@ import ( "fmt" "net" "os" + "os/user" "path/filepath" "strconv" "strings" @@ -22,8 +23,17 @@ type serverAddress struct { // runtime directory selection for console sessions. func default_socket_path() string { if uid := os.Getenv("SUDO_UID"); uid != "" { - if _, err := strconv.ParseUint(uid, 10, 32); err == nil { - return filepath.Join("/run/user", uid, "speech-dispatcher/speechd.sock") + if parsed, err := strconv.ParseUint(uid, 10, 32); err == nil { + 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") diff --git a/internal/speech/autospawn.go b/internal/speech/autospawn.go index 52f14aa4..06f56168 100644 --- a/internal/speech/autospawn.go +++ b/internal/speech/autospawn.go @@ -11,6 +11,8 @@ import ( "net" "os" "os/exec" + "os/user" + "path/filepath" "strconv" "strings" "syscall" @@ -102,10 +104,6 @@ func spawn_arguments(ctx context.Context, endpoint serverAddress) ([]string, err } 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) defer cancel() args, err := spawn_arguments(spawnContext, endpoint) @@ -117,6 +115,24 @@ func spawn_server(ctx context.Context, endpoint serverAddress) error { command = "speech-dispatcher" } 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 output, err := process.CombinedOutput() if err != nil { @@ -124,3 +140,48 @@ func spawn_server(ctx context.Context, endpoint serverAddress) error { } 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 +} diff --git a/internal/speech/autospawn_test.go b/internal/speech/autospawn_test.go index 3314c297..8b537c5b 100644 --- a/internal/speech/autospawn_test.go +++ b/internal/speech/autospawn_test.go @@ -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)) err := spawn_server(context.Background(), serverAddress{"unix", "/tmp/custom.sock"}) if err == nil || !strings.Contains(err.Error(), "invoking user") { diff --git a/internal/speech/catalog.go b/internal/speech/catalog.go index 42e16fd8..b0977f08 100644 --- a/internal/speech/catalog.go +++ b/internal/speech/catalog.go @@ -4,6 +4,7 @@ package speech import ( + "context" "fmt" "os" "sort" @@ -45,7 +46,7 @@ func Discover(options Options) (Catalog, error) { if err != nil { return result, err } - defer w.conn.Close() + defer w.close() deadline := time.Now().Add(4 * time.Second) query := func(command string) ([]string, error) { 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 // names without verifying them. The resulting client has not spoken anything. 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 { return nil, err } @@ -185,7 +192,7 @@ func Prepare(driver string, options Options) (Transport, error) { } if d.Voices == nil { 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") } @@ -219,7 +226,7 @@ func Prepare(driver string, options Options) (Transport, error) { if !languageFound || !voiceFound { 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 @@ -232,7 +239,7 @@ func discover_voices(options Options) ([]Voice, error) { if err != nil { return nil, err } - defer w.conn.Close() + defer w.close() if options.Module != "" { if _, err = w.command("SET SELF OUTPUT_MODULE " + options.Module); err != nil { return nil, err diff --git a/internal/speech/client.go b/internal/speech/client.go index 3a4b3e40..0f14360b 100644 --- a/internal/speech/client.go +++ b/internal/speech/client.go @@ -43,8 +43,20 @@ type Rejection struct{ Code string } func (r *Rejection) Error() string { return "Speech Dispatcher rejected a command: " + r.Code } type wire struct { - conn net.Conn - reader *bufio.Reader + conn net.Conn + 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) { @@ -102,11 +114,14 @@ func connect_options(options Options) (*wire, error) { if err != nil { 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) { - ctx, cancel := context.WithTimeout(context.Background(), 3*commandTimeout) +func connect_endpoint(parent context.Context, endpoint serverAddress, autospawn, systemFallback bool) (*wire, error) { + if parent == nil { + parent = context.Background() + } + ctx, cancel := context.WithTimeout(parent, 3*commandTimeout) defer cancel() dialer := &net.Dialer{Timeout: commandTimeout} dial := serverDialer(dialer.DialContext) @@ -123,7 +138,7 @@ func connect_endpoint(endpoint serverAddress, autospawn, systemFallback bool) (* if err != nil { 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 { @@ -132,11 +147,13 @@ type Client struct { sendMu sync.Mutex controlMu sync.Mutex closeOnce sync.Once + options Options } type Options struct { // Only Prepare enables startup during its prerequisite selection queries. autospawn bool + context context.Context Socket string Rate int Pitch *int @@ -174,15 +191,15 @@ func Open(options Options) (*Client, error) { if err != nil { 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 { 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) { - data.conn.Close() + data.close() if client.control != nil { - client.control.conn.Close() + client.control.close() } return nil, err } @@ -229,13 +246,31 @@ func Open(options Options) (*Client, error) { if endpoint.network == "unix" { 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 { return fail(err) } if _, err := client.control.command(`SET SELF CLIENT_NAME "fenrir:fenrir:control"`); err != nil { 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 } @@ -315,8 +350,16 @@ func (c *Client) Cancel() error { func (c *Client) Close() { c.closeOnce.Do(func() { // Closing the data socket releases a blocked sender before cancellation. - c.data.conn.Close() + c.data.close() 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() + } +} diff --git a/internal/speech/client_test.go b/internal/speech/client_test.go index 6aeaf508..0c185d8b 100644 --- a/internal/speech/client_test.go +++ b/internal/speech/client_test.go @@ -21,7 +21,9 @@ import ( "bufio" "fmt" "net" + "os" "path/filepath" + "strconv" "strings" "sync" "testing" @@ -299,9 +301,10 @@ func TestLiveParametersUseSpeakingClientAndRejectInvalidValues(t *testing.T) { func TestDefaultSocketForSudo(t *testing.T) { t.Setenv("SPEECHD_ADDRESS", "") t.Setenv("SPEECHD_SOCKET", "") - t.Setenv("SUDO_UID", "1000") - t.Setenv("XDG_RUNTIME_DIR", "/run/user/0") - if got, err := resolve_server_address(""); err != nil || got.address != "/run/user/1000/speech-dispatcher/speechd.sock" { + runtimeDirectory := t.TempDir() + t.Setenv("SUDO_UID", strconv.Itoa(os.Getuid())) + 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.Setenv("SPEECHD_ADDRESS", "unix_socket:/tmp/example-speechd.sock") diff --git a/internal/speech/drivers.go b/internal/speech/drivers.go index 928cdac4..64234570 100644 --- a/internal/speech/drivers.go +++ b/internal/speech/drivers.go @@ -17,7 +17,10 @@ package speech -import "fmt" +import ( + "context" + "fmt" +) // 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. @@ -83,6 +86,12 @@ func Find_driver(id string) (Driver, 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) if err != nil { return nil, err @@ -93,7 +102,11 @@ func Open_driver(id string, options Options) (Transport, error) { if err = driver.Availability(); err != nil { 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 @@ -103,7 +116,7 @@ func discover_modules(options Options) ([]string, error) { if err != nil { return nil, err } - defer wire.conn.Close() + defer wire.close() modules, err := wire.command("LIST OUTPUT_MODULES") if err != nil { return nil, err diff --git a/internal/speech/recovery_test.go b/internal/speech/recovery_test.go new file mode 100644 index 00000000..466366dd --- /dev/null +++ b/internal/speech/recovery_test.go @@ -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") +} diff --git a/internal/speech/service.go b/internal/speech/service.go index 596cc3c0..36aaaf36 100644 --- a/internal/speech/service.go +++ b/internal/speech/service.go @@ -18,7 +18,11 @@ package speech import ( + "context" "errors" + "log/slog" + + "git.stormux.org/storm/fenrir/internal/audio" "strings" "sync" "time" @@ -39,6 +43,11 @@ type Transport interface { type Service struct { transport Transport transportMu sync.RWMutex + transportEpoch uint64 + faults chan transportFailure + recoveryContext context.Context + stopRecovery context.CancelFunc + recovering bool replacements []replacement mu sync.Mutex dispatch sync.Mutex @@ -52,6 +61,11 @@ type Service struct { workers sync.WaitGroup } +type transportFailure struct { + epoch uint64 + err error +} + type replacement struct { transport Transport done chan error @@ -82,7 +96,8 @@ type utterance struct { } 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) go s.deliver() go s.cancel() @@ -143,7 +158,7 @@ func (s *Service) enqueue(item utterance) { } s.mu.Lock() defer s.mu.Unlock() - if s.closed || s.muted { + if s.closed || s.muted || s.recovering { return } 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() { defer s.workers.Done() normalPitch := 0 @@ -198,6 +279,10 @@ func (s *Service) deliver() { case <-s.done: return case <-s.wake: + case fault := <-s.faults: + if fault.epoch == s.transportEpoch && !s.recover_transport(fault.err, currentParameters) { + return + } } for { s.mu.Lock() @@ -211,8 +296,9 @@ func (s *Service) deliver() { err := s.transport.Configure(parameters) s.dispatch.Unlock() if err != nil { - s.fail(err) - return + if !s.recover_transport(err, currentParameters) { + return + } } continue } @@ -233,6 +319,7 @@ func (s *Service) deliver() { s.transportMu.Lock() old := s.transport s.transport = request.transport + s.transportEpoch++ s.transportMu.Unlock() s.mu.Unlock() old.Close() @@ -262,8 +349,9 @@ func (s *Service) deliver() { diagnostic.Timing("speech send", started, "generation", generation) s.dispatch.Unlock() if err != nil { - s.fail(err) - return + if !s.recover_transport(err, currentParameters) { + return + } } } } @@ -293,7 +381,17 @@ func (s *Service) say(item utterance, normalPitch int, normalParameters *Paramet func (s *Service) cancel_transport() error { s.transportMu.RLock() 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() { @@ -320,21 +418,13 @@ func (s *Service) cancel() { if !s.dispatch.TryLock() { // Stop audible speech even while a stale send is blocked. A // second cancel after that send finishes closes the late-send race. - if err := s.cancel_transport(); err != nil { - s.fail(err) - return - } + _ = s.cancel_transport() s.dispatch.Lock() } // With dispatch held, no send can race this cancel. The usual // idle-send path needs only this one acknowledgement. - err := s.cancel_transport() + _ = s.cancel_transport() diagnostic.Timing("speech cancel barrier", started, "generation", generation) - if err != nil { - s.dispatch.Unlock() - s.fail(err) - return - } s.mu.Lock() // Every intervening interrupt is covered: fresh sends have been // gated throughout. Coalesce them instead of canceling again. @@ -354,6 +444,7 @@ func (s *Service) Close() { return } s.closed = true + s.stopRecovery() s.queue = nil close(s.done) pending := s.replacements @@ -364,7 +455,9 @@ func (s *Service) Close() { request.done <- errors.New("speech service closed") } s.transportMu.RLock() - s.transport.Close() + if s.transport != nil { + s.transport.Close() + } s.transportMu.RUnlock() s.workers.Wait() }