2 Commits
Author SHA1 Message Date
kato b0487be39e feat: implement isolated model worker with auto-restart and hot-reload
Tests / Go and GTK tests (push) Successful in 1m38s
Release / Tests before release (push) Successful in 1m34s
Release / Build and publish release (push) Successful in 2m24s
Introduce a supervised child process for capture, VAD, and recognition,
enabling automatic restart on crash/hang and hot-reload via SIGHUP.
Add `--model-auto-restart`, `--model-timeout`, and
`--model-shutdown-timeout` options. Refactor `Run()` with lifecycle
hooks for proper process group cleanup. Update README with new
behavior and CLI flags.
2026-07-17 01:47:37 +03:00
kato b3ed538820 ci(release): add patchelf to CI workflows and update release documentation
Tests / Go and GTK tests (push) Canceled after 40s
Release / Tests before release (push) Successful in 1m44s
Release / Build and publish release (push) Successful in 3m2s
- Install patchelf in both build and release jobs to support runtime library relocation.
- Update release notes to describe the private desktop runtime archive.
- Clarify in README that development builds require GTK4 dev libraries, while release archives are self-contained with lib/ directory.
- Adjust troubleshooting and release procedure to reflect new offline runtime packaging.
2026-07-16 22:09:38 +03:00
13 changed files with 864 additions and 31 deletions
+3 -3
View File
@@ -27,7 +27,7 @@ jobs:
run: >-
dnf install -y
bzip2 curl gcc gcc-c++ git gtk4-devel gtk4-layer-shell-devel
jq libX11-devel libXfixes-devel nodejs24 nodejs24-bin pkgconf-pkg-config
jq libX11-devel libXfixes-devel nodejs24 nodejs24-bin patchelf pkgconf-pkg-config
pulseaudio-utils ShellCheck tar gzip binutils
- name: Check out repository
@@ -63,7 +63,7 @@ jobs:
run: >-
dnf install -y
bzip2 curl gcc gcc-c++ git gtk4-devel gtk4-layer-shell-devel
jq libX11-devel libXfixes-devel nodejs24 nodejs24-bin pkgconf-pkg-config
jq libX11-devel libXfixes-devel nodejs24 nodejs24-bin patchelf pkgconf-pkg-config
pulseaudio-utils ShellCheck tar gzip binutils
- name: Check out repository and tags
@@ -95,7 +95,7 @@ jobs:
previous_tag="$(git describe --tags --abbrev=0 "${GITHUB_REF_NAME}^" --match 'v[0-9]*' 2>/dev/null || true)"
{
printf 'Automated Captioneer release for `%s`.\n\n' "${GITHUB_REF_NAME}"
printf 'The Linux x86_64 archive contains both Captioneer commands and their Sherpa/ONNX native libraries. Models are downloaded separately with `scripts/download-models.sh`.\n'
printf 'The Linux x86_64 archive contains both Captioneer commands and a private desktop runtime with GTK4, GTK4 Layer Shell, and Sherpa/ONNX libraries. Models are downloaded separately with `scripts/download-models.sh`.\n'
if [[ -n "$previous_tag" ]]; then
printf '\nChanges since `%s`:\n\n' "$previous_tag"
git log --no-merges --pretty='- %s (`%h`)' "${previous_tag}..${GITHUB_REF_NAME}"
+17 -5
View File
@@ -11,7 +11,7 @@ Captioneer provides two commands:
## Setup
The terminal command requires Linux x86_64, Go 1.26.2 or newer, `pactl`, and `parec`. The desktop command additionally needs GTK4 and GTK4 Layer Shell development/runtime libraries. PipeWire works through its PulseAudio compatibility layer.
The terminal command requires Linux x86_64, Go 1.26.2 or newer, `pactl`, and `parec`. Building the desktop command from source additionally needs GTK4 and GTK4 Layer Shell development libraries. Published release archives carry the desktop runtime libraries. PipeWire works through its PulseAudio compatibility layer.
Download the large recognition models once:
@@ -50,6 +50,14 @@ Final captions use stdout; diagnostics and overload warnings use stderr. This ma
Press <kbd>Ctrl</kbd>+<kbd>C</kbd> to stop. Captioneer stops `parec` and flushes pending audio before exiting.
Sherpa recognition, Silero VAD, and audio capture run in an isolated worker process. The terminal or GTK process stays responsive if native inference crashes or hangs: Captioneer restarts the worker automatically after a failure, and Ctrl+C force-kills only that worker if it cannot flush within the configured shutdown timeout. Send `SIGHUP` to the parent Captioneer PID to hot-reload the models without closing the terminal or overlay:
```bash
kill -HUP <captioneer-parent-pid>
```
An automatic or manual restart begins a fresh capture/transcription session; audio being processed by the failed worker cannot be recovered.
## Desktop overlay
The simplest Bazzite development command is:
@@ -98,6 +106,9 @@ Layer Shell and X11 hints are requests, not universal guarantees. Same-layer win
| `--models-dir=models` | `models` | Directory containing `parakeet-tdt-v2/` and `silero_vad_v5.onnx`. |
| `--threads=2` | `2` | CPU threads used by recognition and VAD. |
| `--preview-threshold-dbfs=-45` | `-45` | RMS threshold that starts a provisional VAD caption. Raise it to reject background audio; lower it for quiet speech. |
| `--model-auto-restart=true` | `true` | Restart the isolated model worker after a crash, capture failure, or timeout. Set `false` to report the failure and exit instead. |
| `--model-timeout=30s` | `30s` | Maximum model startup, continuously busy processing time, or missing-heartbeat period before the worker is considered hung. `0s` disables hang detection. |
| `--model-shutdown-timeout=2s` | `2s` | Time allowed for pending-audio flush and model cleanup after Ctrl+C or reload before the worker is force-killed. |
### Desktop-only options
@@ -165,11 +176,12 @@ Captioneer itself has no environment-based application configuration; use the co
- **Processing overload warning:** inference fell behind and the oldest 100 ms packet was dropped. Try more `--threads` or a longer `--chunk-duration`.
- **Overlay is not always above other windows:** read the backend warning. On Wayland this usually means the compositor lacks Layer Shell; add a compositor rule for the `Captioneer Overlay` window if supported.
- **Overlay initialization fails:** `--output=overlay` exits with the error. `--output=both` warns and continues terminal-only.
- **Desktop binary misses shared libraries on Bazzite:** use `scripts/distrobox/run.sh`; container-built GTK binaries are not self-contained.
- **A locally built desktop binary misses shared libraries:** use `scripts/distrobox/run.sh` or install the matching GTK4 and GTK4 Layer Shell runtime packages. The desktop command in a published release carries these application libraries in its private `lib/` directory; keep the extracted directory together.
- **The model worker repeatedly restarts:** inspect the warning on stderr. Slow machines may need a larger `--model-timeout`; use `--model-auto-restart=false` when diagnosing a persistent crash so Captioneer exits on the first failure.
## Development
The transcription pipeline emits presentation-neutral events. `output/terminal` and `output/overlay` consume those events independently; GTK stays on its main OS thread while capture and recognition run on a worker.
The transcription pipeline emits presentation-neutral events. `output/terminal` and `output/overlay` consume those events independently; GTK stays on its main OS thread while capture, VAD, and recognition run in a supervised child process. Caption events and health messages cross a private pipe, and the child is placed in its own process group so its `parec` process can be stopped or replaced with it.
```text
src/app/ capture/model/process lifecycle
@@ -212,12 +224,12 @@ git tag v1.3.519
git push origin v1.3.519
```
The release contains a checksummed Linux x86_64 archive with `captioneer`, `captioneer-desktop`, the required Sherpa/ONNX shared libraries, the model download script, the README, and the license. The desktop command still requires GTK4 and GTK4 Layer Shell runtime libraries on the destination system.
The release contains a checksummed Linux x86_64 archive with `captioneer`, `captioneer-desktop`, and a private runtime containing GTK4, GTK4 Layer Shell, Sherpa/ONNX, and their linked non-glibc dependencies. Extract the whole archive and run `./captioneer-desktop`; do not move the executable away from its adjacent `lib/` directory. Models, `pactl`, `parec`, host fonts, the display server, graphics drivers, and glibc remain host-provided.
The release workflow uses Gitea's built-in `GITEA_TOKEN`; no personal token is required. In **Settings → Actions → General**, allow the workflow token `Read` access to code and `Write` access to releases. Gitea clamps workflow permissions to those repository settings, so a more restrictive maximum produces an actionable HTTP 403 during publishing.
## License
Captioneer is licensed under the [Apache License 2.0](LICENSE). Sherpa-Onnx is Apache-2.0 licensed, GTK4 is LGPL-2.1-or-later, and GTK4 Layer Shell is MIT licensed. Captioneer links to the system GTK libraries rather than copying their source into the project.
Captioneer is licensed under the [Apache License 2.0](LICENSE). Sherpa-Onnx is Apache-2.0 licensed, GTK4 is LGPL-2.1-or-later, and GTK4 Layer Shell is MIT licensed. Release archives dynamically link and redistribute the required shared libraries; `RUNTIME-LIBRARIES.tsv` records their source packages and `third-party-licenses/` contains the installed license notices supplied by those packages. Local source builds continue to link against the system libraries.
Version 1 intentionally has no animation, persisted caption history, runtime dragging, background-free mode, capture exclusion, Flatpak, desktop entry, or autostart integration.
+176 -9
View File
@@ -4,6 +4,13 @@ set -Eeuo pipefail
repo_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd -P)"
cd -- "$repo_root"
for command in go ldd patchelf readelf rpm; do
command -v "$command" >/dev/null 2>&1 || {
printf 'captioneer: %s is required to build a release archive\n' "$command" >&2
exit 1
}
done
tag=${1:-}
if [[ ! "$tag" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
printf 'captioneer: release tag must match vMAJOR.MINOR.BUG, got %q\n' "$tag" >&2
@@ -17,29 +24,189 @@ package_dir="dist/$package_name"
rm -rf -- "$package_dir"
rm -f -- "dist/$archive_name" "dist/$archive_name.sha256"
mkdir -p -- "$package_dir/lib" "$package_dir/scripts"
mkdir -p -- \
"$package_dir/lib" \
"$package_dir/scripts" \
"$package_dir/third-party-licenses"
# The Sherpa Go module embeds an absolute build-machine RPATH. Add a portable
# fallback so the bundled native libraries resolve beside the extracted tools.
# build-time fallback; the final paths are normalized after dependency copying.
# shellcheck disable=SC2016 # $ORIGIN must remain literal for the ELF loader.
export CGO_LDFLAGS='-Wl,-rpath,$ORIGIN/lib'
go build -trimpath -o "$package_dir/captioneer" ./src/cmd/captioneer
go build -trimpath -tags gtk -o "$package_dir/captioneer-desktop" ./src/cmd/captioneer-desktop
CGO_LDFLAGS='-Wl,-rpath,$ORIGIN/lib' \
go build -trimpath -o "$package_dir/captioneer" ./src/cmd/captioneer
# shellcheck disable=SC2016 # $ORIGIN must remain literal for the ELF loader.
CGO_LDFLAGS='-Wl,-rpath,$ORIGIN/lib' \
go build -trimpath -tags gtk \
-o "$package_dir/captioneer-desktop" ./src/cmd/captioneer-desktop
sherpa_dir="$(go list -m -f '{{.Dir}}' github.com/k2-fsa/sherpa-onnx-go-linux)"
sherpa_lib_dir="$sherpa_dir/lib/x86_64-unknown-linux-gnu"
cp -- "$sherpa_lib_dir/libsherpa-onnx-c-api.so" "$package_dir/lib/"
cp -- "$sherpa_lib_dir/libonnxruntime.so" "$package_dir/lib/"
declare -A bundled_libraries=()
declare -A copied_license_packages=()
manifest_entries="$(mktemp)"
trap 'rm -f -- "$manifest_entries"' EXIT
is_host_runtime_library() {
case "$1" in
ld-linux*.so* | libc.so.* | libm.so.* | libmvec.so.* | libdl.so.* | \
libpthread.so.* | librt.so.* | libresolv.so.* | libanl.so.* | \
libutil.so.* | libnss_*.so.* | libthread_db.so.* | libBrokenLocale.so.*)
return 0
;;
esac
return 1
}
copy_package_licenses() {
local package=$1
local license_path relative_path destination
[[ -z "${copied_license_packages[$package]:-}" ]] || return 0
copied_license_packages["$package"]=1
while IFS= read -r license_path; do
[[ -f "$license_path" && "$license_path" == /usr/share/licenses/* ]] || continue
relative_path="${license_path#/usr/share/licenses/}"
destination="$package_dir/third-party-licenses/$relative_path"
mkdir -p -- "$(dirname -- "$destination")"
cp -L -- "$license_path" "$destination"
done < <(rpm -ql "$package" 2>/dev/null || true)
}
bundle_library() {
local source=$1
local provider=${2:-}
local library_name resolved_source destination package
library_name="$(basename -- "$source")"
is_host_runtime_library "$library_name" && return
resolved_source="$(readlink -f -- "$source")"
[[ -f "$resolved_source" ]] || {
printf 'captioneer: runtime library not found: %s\n' "$source" >&2
exit 1
}
if [[ -n "${bundled_libraries[$library_name]:-}" ]]; then
if ! cmp -s -- "$resolved_source" "${bundled_libraries[$library_name]}"; then
printf 'captioneer: conflicting runtime libraries named %s\n' "$library_name" >&2
exit 1
fi
return
fi
destination="$package_dir/lib/$library_name"
cp -L -- "$resolved_source" "$destination"
bundled_libraries["$library_name"]="$resolved_source"
if [[ -z "$provider" ]] && command -v rpm >/dev/null 2>&1; then
package="$(rpm -qf --queryformat '%{NAME}\n' "$resolved_source" 2>/dev/null | head -n 1 || true)"
if [[ -n "$package" ]]; then
provider="$(rpm -q --queryformat '%{NAME}-%{VERSION}-%{RELEASE}.%{ARCH}' "$package")"
copy_package_licenses "$package"
fi
fi
provider=${provider:-unidentified-system-package}
printf '%s\t%s\n' "$library_name" "$provider" >> "$manifest_entries"
}
bundle_library "$sherpa_lib_dir/libsherpa-onnx-c-api.so" \
'github.com/k2-fsa/sherpa-onnx-go-linux'
bundle_library "$sherpa_lib_dir/libonnxruntime.so" \
'github.com/k2-fsa/sherpa-onnx-go-linux'
# ldd reports the complete linked dependency closure. Copy everything needed by
# either command except glibc and its loader, which must match the host system.
for executable in \
"$package_dir/captioneer" \
"$package_dir/captioneer-desktop"; do
ldd_output="$(ldd "$executable")"
if grep -Fq 'not found' <<< "$ldd_output"; then
printf 'captioneer: unresolved release dependencies for %s:\n%s\n' \
"$executable" "$ldd_output" >&2
exit 1
fi
while IFS= read -r library_path; do
[[ -n "$library_path" ]] || continue
if [[ "$library_path" == "$repo_root/$package_dir/lib/"* ]]; then
continue
fi
bundle_library "$library_path"
done < <(
awk '
$2 == "=>" && $3 ~ /^\// { print $3 }
$1 ~ /^\// { print $1 }
' <<< "$ldd_output"
)
done
{
printf 'Library\tProvider\n'
sort -u -- "$manifest_entries"
} > "$package_dir/RUNTIME-LIBRARIES.tsv"
# Give every object a location-relative path. Patching each library is required
# because an upstream RUNPATH (for example, Tinysparql's Fedora path) otherwise
# prevents the executable's path from resolving that library's own children.
# This avoids LD_LIBRARY_PATH, which would leak into spawned pactl/parec tools.
for library in "$package_dir"/lib/*; do
chmod u+w -- "$library"
# shellcheck disable=SC2016 # $ORIGIN must remain literal for the ELF loader.
patchelf --set-rpath '$ORIGIN' "$library"
chmod 0755 -- "$library"
done
for executable in captioneer captioneer-desktop; do
# shellcheck disable=SC2016 # $ORIGIN must remain literal for the ELF loader.
patchelf --set-rpath '$ORIGIN/lib' "$package_dir/$executable"
done
cp -- README.md LICENSE "$package_dir/"
cp -- scripts/download-models.sh "$package_dir/scripts/"
for executable in captioneer captioneer-desktop; do
dynamic_section="$(readelf -d "$package_dir/$executable")"
# shellcheck disable=SC2016 # Verify the literal ELF-loader token.
if ! readelf -d "$package_dir/$executable" | grep -Fq '$ORIGIN/lib'; then
printf 'captioneer: %s is missing the portable native-library RPATH\n' "$executable" >&2
if [[ "$dynamic_section" != *'$ORIGIN/lib'* ]]; then
printf 'captioneer: %s is missing its private-runtime RUNPATH\n' \
"$executable" >&2
exit 1
fi
done
[[ -f "$package_dir/lib/libgtk4-layer-shell.so.0" ]] || {
printf 'captioneer: GTK4 Layer Shell was not included in the release runtime\n' >&2
exit 1
}
for library in "$package_dir"/lib/*; do
# shellcheck disable=SC2016 # Verify the literal ELF-loader token.
[[ "$(patchelf --print-rpath "$library")" == '$ORIGIN' ]] || {
printf 'captioneer: %s is missing its private dependency RUNPATH\n' \
"$(basename -- "$library")" >&2
exit 1
}
done
# Verify the archive itself contains every linked non-glibc dependency instead
# of allowing the build container's installed libraries to mask an omission.
for object in \
"$package_dir/captioneer" \
"$package_dir/captioneer-desktop" \
"$package_dir"/lib/*; do
while IFS= read -r needed_library; do
is_host_runtime_library "$needed_library" && continue
[[ -f "$package_dir/lib/$needed_library" ]] || {
printf 'captioneer: %s requires unbundled library %s\n' \
"$(basename -- "$object")" "$needed_library" >&2
exit 1
}
done < <(patchelf --print-needed "$object")
done
# --help exits before display/model initialization while still exercising the
# dynamic loader through the same executable release users invoke.
"$package_dir/captioneer-desktop" --help >/dev/null 2>&1
tar -C dist -czf "dist/$archive_name" "$package_name"
(
+35 -5
View File
@@ -14,15 +14,28 @@ import (
"tea.chunkbyte.com/kato/captioneer/src/output"
)
func Run(ctx context.Context, settings config.Settings, sink output.Sink, diagnostics io.Writer) error {
if err := capture.ValidatePrograms(); err != nil {
type directHooks struct {
ready func() error
busy func() error
idle func() error
}
func runDirect(ctx context.Context, settings config.Settings, sink output.Sink, diagnostics io.Writer, hooks directHooks) error {
if err := hooks.busy(); err != nil {
return err
}
resources, err := models.New(settings)
if idleErr := hooks.idle(); err == nil {
err = idleErr
}
if err != nil {
return err
}
defer resources.Close()
defer func() {
_ = hooks.busy()
resources.Close()
_ = hooks.idle()
}()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
@@ -32,24 +45,41 @@ func Run(ctx context.Context, settings config.Settings, sink output.Sink, diagno
return fmt.Errorf("find default output monitor: %w", err)
}
fmt.Fprintf(diagnostics, "Capturing from: %s\n", monitorSource)
fmt.Fprintln(diagnostics, "Press Ctrl+C to stop.")
packets, waitCapture, err := capture.Packets(ctx, monitorSource)
if err != nil {
return fmt.Errorf("start system-audio capture: %w", err)
}
if err := hooks.ready(); err != nil {
cancel()
_ = waitCapture()
return err
}
processor := captions.New(settings, resources, sink.Publish)
var processErr error
for packet := range packets {
if err := processor.Accept(packet); err != nil {
if err := hooks.busy(); err != nil {
processErr = err
cancel()
break
}
acceptErr := processor.Accept(packet)
idleErr := hooks.idle()
if acceptErr != nil || idleErr != nil {
err := errors.Join(acceptErr, idleErr)
processErr = fmt.Errorf("publish caption: %w", err)
cancel()
break
}
}
if processErr == nil {
if err := hooks.busy(); err != nil {
processErr = err
} else {
processErr = processor.Flush()
processErr = errors.Join(processErr, hooks.idle())
}
}
captureErr := waitCapture()
if captureErr != nil && ctx.Err() == nil {
+361
View File
@@ -0,0 +1,361 @@
package app
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/exec"
"os/signal"
"strings"
"syscall"
"time"
"tea.chunkbyte.com/kato/captioneer/src/capture"
"tea.chunkbyte.com/kato/captioneer/src/config"
"tea.chunkbyte.com/kato/captioneer/src/models"
"tea.chunkbyte.com/kato/captioneer/src/output"
)
const modelRestartDelay = time.Second
type attemptResult uint8
const (
attemptStopped attemptResult = iota + 1
attemptReload
attemptFailed
attemptFatal
)
type workerHealth struct {
startedAt time.Time
lastSeen time.Time
ready bool
busySince time.Time
}
func (h *workerHealth) observe(messageType string, now time.Time) {
h.lastSeen = now
switch messageType {
case messageReady:
h.ready = true
h.busySince = time.Time{}
case messageBusy:
if h.busySince.IsZero() {
h.busySince = now
}
case messageIdle:
h.busySince = time.Time{}
}
}
func (h workerHealth) timeoutError(now time.Time, timeout time.Duration) error {
if timeout == 0 {
return nil
}
if !h.ready && now.Sub(h.startedAt) >= timeout {
return fmt.Errorf("model worker did not become ready within %s", timeout)
}
if !h.busySince.IsZero() && now.Sub(h.busySince) >= timeout {
return fmt.Errorf("model operation exceeded %s", timeout)
}
if h.ready && now.Sub(h.lastSeen) >= timeout {
return fmt.Errorf("model worker heartbeat was silent for %s", timeout)
}
return nil
}
// Run supervises an isolated process containing capture, VAD, and recognition.
// Presentation stays in this process, so the native model can be replaced
// without restarting the terminal or GTK main loop.
func Run(ctx context.Context, settings config.Settings, sink output.Sink, diagnostics io.Writer) error {
if err := capture.ValidatePrograms(); err != nil {
return err
}
if err := models.Validate(settings); err != nil {
return err
}
fmt.Fprintln(diagnostics, "Press Ctrl+C to stop. Send SIGHUP to hot-reload the model worker.")
reload := make(chan os.Signal, 1)
signal.Notify(reload, syscall.SIGHUP)
defer signal.Stop(reload)
for {
result, err := runWorkerAttempt(ctx, settings, sink, diagnostics, reload)
if ctx.Err() != nil {
return nil
}
switch result {
case attemptStopped:
return nil
case attemptFatal:
return err
case attemptReload:
fmt.Fprintln(diagnostics, "Reloading model worker.")
continue
case attemptFailed:
if !settings.ModelAutoRestart {
return err
}
fmt.Fprintf(diagnostics, "warning: %v; restarting model worker in %s\n", err, modelRestartDelay)
}
timer := time.NewTimer(modelRestartDelay)
select {
case <-ctx.Done():
if !timer.Stop() {
<-timer.C
}
return nil
case <-reload:
if !timer.Stop() {
<-timer.C
}
fmt.Fprintln(diagnostics, "Reloading model worker.")
case <-timer.C:
}
}
}
func runWorkerAttempt(
ctx context.Context,
settings config.Settings,
sink output.Sink,
diagnostics io.Writer,
reload <-chan os.Signal,
) (attemptResult, error) {
cmd, protocol, err := startModelWorker(settings, diagnostics)
if err != nil {
return attemptFailed, fmt.Errorf("start model worker: %w", err)
}
defer protocol.Close()
messages := make(chan workerMessage)
protocolDone := make(chan error, 1)
go readWorkerMessages(protocol, messages, protocolDone)
wait := make(chan error, 1)
go func() { wait <- cmd.Wait() }()
checkInterval := 250 * time.Millisecond
if settings.ModelTimeout > 0 && settings.ModelTimeout < checkInterval {
checkInterval = settings.ModelTimeout / 2
if checkInterval <= 0 {
checkInterval = time.Millisecond
}
}
ticker := time.NewTicker(checkInterval)
defer ticker.Stop()
startedAt := time.Now()
health := workerHealth{startedAt: startedAt, lastSeen: startedAt}
messageChannel := (<-chan workerMessage)(messages)
waitChannel := (<-chan error)(wait)
contextDone := ctx.Done()
var stopTimer *time.Timer
var stopDeadline <-chan time.Time
var stopping bool
var desiredResult attemptResult
var desiredErr error
var processDone bool
var processErr error
var protocolClosed bool
var protocolErr error
var workerFailure string
beginStop := func(result attemptResult, stopErr error) {
if stopping {
if result == attemptFatal {
desiredResult = result
desiredErr = errors.Join(desiredErr, stopErr)
}
return
}
stopping = true
desiredResult = result
desiredErr = stopErr
if err := signalWorkerGroup(cmd, syscall.SIGTERM); err != nil {
desiredErr = errors.Join(desiredErr, err)
}
stopTimer = time.NewTimer(settings.ModelShutdownTimeout)
stopDeadline = stopTimer.C
}
for {
if processDone && protocolClosed {
if stopTimer != nil && !stopTimer.Stop() {
select {
case <-stopTimer.C:
default:
}
}
if stopping {
return desiredResult, desiredErr
}
// The worker may have crashed before its parec child. A process group
// remains signalable after its leader exits, so clean up any survivors
// before starting a replacement worker.
cleanupErr := signalWorkerGroup(cmd, syscall.SIGKILL)
if protocolErr != nil {
return attemptFailed, errors.Join(fmt.Errorf("model worker protocol: %w", protocolErr), cleanupErr)
}
if workerFailure != "" {
return attemptFailed, errors.Join(errors.New(workerFailure), cleanupErr)
}
if processErr != nil {
return attemptFailed, errors.Join(fmt.Errorf("model worker exited: %w", processErr), cleanupErr)
}
return attemptFailed, errors.Join(errors.New("model worker exited unexpectedly"), cleanupErr)
}
select {
case <-contextDone:
contextDone = nil
beginStop(attemptStopped, nil)
case <-reload:
beginStop(attemptReload, nil)
case message, ok := <-messageChannel:
if !ok {
messageChannel = nil
protocolClosed = true
protocolErr = <-protocolDone
if !processDone && !stopping {
beginStop(attemptFailed, errors.New("model worker protocol closed unexpectedly"))
}
continue
}
now := time.Now()
health.observe(message.Type, now)
switch message.Type {
case messageReady:
fmt.Fprintln(diagnostics, "Model worker ready.")
case messageBusy, messageIdle:
case messageHeartbeat:
case messageCaption:
if message.Event == nil {
beginStop(attemptFailed, errors.New("model worker sent an empty caption event"))
continue
}
if err := sink.Publish(*message.Event); err != nil {
beginStop(attemptFatal, fmt.Errorf("publish caption: %w", err))
}
case messageFailure:
workerFailure = message.Error
default:
beginStop(attemptFailed, fmt.Errorf("unknown model worker message %q", message.Type))
}
case err := <-waitChannel:
waitChannel = nil
processDone = true
processErr = err
case now := <-ticker.C:
if !stopping {
if err := health.timeoutError(now, settings.ModelTimeout); err != nil {
beginStop(attemptFailed, err)
}
}
case <-stopDeadline:
stopDeadline = nil
if !processDone {
fmt.Fprintf(diagnostics, "warning: model worker did not stop within %s; force-killing it\n", settings.ModelShutdownTimeout)
if err := signalWorkerGroup(cmd, syscall.SIGKILL); err != nil {
desiredErr = errors.Join(desiredErr, err)
}
}
}
}
}
func startModelWorker(settings config.Settings, diagnostics io.Writer) (*exec.Cmd, *os.File, error) {
executable, err := os.Executable()
if err != nil {
return nil, nil, err
}
encodedSettings, err := encodeWorkerSettings(settings)
if err != nil {
return nil, nil, err
}
protocolRead, protocolWrite, err := os.Pipe()
if err != nil {
return nil, nil, err
}
cmd := exec.Command(executable)
cmd.Env = workerEnvironment(encodedSettings)
cmd.ExtraFiles = []*os.File{protocolWrite}
cmd.Stdout = diagnostics
cmd.Stderr = diagnostics
cmd.SysProcAttr = &syscall.SysProcAttr{
Setpgid: true,
Pdeathsig: syscall.SIGKILL,
}
if err := cmd.Start(); err != nil {
protocolRead.Close()
protocolWrite.Close()
return nil, nil, err
}
if err := protocolWrite.Close(); err != nil {
_ = signalWorkerGroup(cmd, syscall.SIGKILL)
_ = cmd.Wait()
protocolRead.Close()
return nil, nil, err
}
return cmd, protocolRead, nil
}
func workerEnvironment(encodedSettings string) []string {
environment := make([]string, 0, len(os.Environ())+2)
for _, item := range os.Environ() {
if strings.HasPrefix(item, modelWorkerEnvironment+"=") ||
strings.HasPrefix(item, modelSettingsEnvironment+"=") {
continue
}
environment = append(environment, item)
}
return append(environment,
modelWorkerEnvironment+"=1",
modelSettingsEnvironment+"="+encodedSettings,
)
}
func readWorkerMessages(protocol io.Reader, messages chan<- workerMessage, done chan<- error) {
defer close(messages)
decoder := json.NewDecoder(protocol)
for {
var message workerMessage
if err := decoder.Decode(&message); err != nil {
if errors.Is(err, io.EOF) {
done <- nil
} else {
done <- err
}
return
}
messages <- message
}
}
func signalWorkerGroup(cmd *exec.Cmd, signal syscall.Signal) error {
if cmd.Process == nil {
return nil
}
err := syscall.Kill(-cmd.Process.Pid, signal)
if errors.Is(err, syscall.ESRCH) {
return nil
}
if err != nil {
return fmt.Errorf("signal model worker: %w", err)
}
return nil
}
+66
View File
@@ -0,0 +1,66 @@
package app
import (
"strings"
"testing"
"time"
"tea.chunkbyte.com/kato/captioneer/src/config"
)
func TestWorkerHealthDetectsStartupAndOperationTimeouts(t *testing.T) {
started := time.Unix(100, 0)
health := workerHealth{startedAt: started, lastSeen: started}
if err := health.timeoutError(started.Add(time.Second), 2*time.Second); err != nil {
t.Fatalf("healthy startup reported a timeout: %v", err)
}
if err := health.timeoutError(started.Add(2*time.Second), 2*time.Second); err == nil ||
!strings.Contains(err.Error(), "did not become ready") {
t.Fatalf("startup timeout = %v", err)
}
health.observe(messageReady, started.Add(2*time.Second))
health.observe(messageBusy, started.Add(3*time.Second))
if err := health.timeoutError(started.Add(4*time.Second), 2*time.Second); err != nil {
t.Fatalf("healthy operation reported a timeout: %v", err)
}
if err := health.timeoutError(started.Add(5*time.Second), 2*time.Second); err == nil ||
!strings.Contains(err.Error(), "operation exceeded") {
t.Fatalf("operation timeout = %v", err)
}
health.observe(messageIdle, started.Add(5*time.Second))
health.observe(messageHeartbeat, started.Add(6*time.Second))
if err := health.timeoutError(started.Add(7*time.Second), 2*time.Second); err != nil {
t.Fatalf("idle worker reported a timeout: %v", err)
}
if err := health.timeoutError(started.Add(8*time.Second), 2*time.Second); err == nil ||
!strings.Contains(err.Error(), "heartbeat was silent") {
t.Fatalf("heartbeat timeout = %v", err)
}
}
func TestWorkerSettingsRoundTrip(t *testing.T) {
want := config.Settings{
Mode: config.ModeVAD,
ChunkDuration: time.Second,
ModelsDir: "models with spaces",
Threads: 3,
PreviewThresholdDB: -42,
ModelAutoRestart: true,
ModelTimeout: 12 * time.Second,
ModelShutdownTimeout: 2 * time.Second,
}
encoded, err := encodeWorkerSettings(want)
if err != nil {
t.Fatal(err)
}
got, err := decodeWorkerSettings(encoded)
if err != nil {
t.Fatal(err)
}
if got != want {
t.Fatalf("settings round trip = %#v, want %#v", got, want)
}
}
+152
View File
@@ -0,0 +1,152 @@
package app
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/signal"
"sync"
"syscall"
"time"
"tea.chunkbyte.com/kato/captioneer/src/captions"
"tea.chunkbyte.com/kato/captioneer/src/config"
)
const (
modelWorkerEnvironment = "CAPTIONEER_INTERNAL_MODEL_WORKER"
modelSettingsEnvironment = "CAPTIONEER_INTERNAL_MODEL_SETTINGS"
modelProtocolFD = 3
)
const (
messageReady = "ready"
messageBusy = "busy"
messageIdle = "idle"
messageCaption = "caption"
messageFailure = "failure"
messageHeartbeat = "heartbeat"
)
type workerMessage struct {
Type string `json:"type"`
Event *captions.Event `json:"event,omitempty"`
Error string `json:"error,omitempty"`
}
type workerReporter struct {
mu sync.Mutex
encoder *json.Encoder
}
func (r *workerReporter) send(message workerMessage) error {
r.mu.Lock()
defer r.mu.Unlock()
return r.encoder.Encode(message)
}
type protocolSink struct {
reporter *workerReporter
}
func (s protocolSink) Publish(event captions.Event) error {
return s.reporter.send(workerMessage{Type: messageCaption, Event: &event})
}
func (protocolSink) Close() error { return nil }
// IsModelWorker reports whether this process was created as Captioneer's
// private inference worker. The environment marker is intentionally internal.
func IsModelWorker() bool {
return os.Getenv(modelWorkerEnvironment) == "1"
}
func encodeWorkerSettings(settings config.Settings) (string, error) {
data, err := json.Marshal(settings)
if err != nil {
return "", err
}
return base64.RawStdEncoding.EncodeToString(data), nil
}
func decodeWorkerSettings(value string) (config.Settings, error) {
data, err := base64.RawStdEncoding.DecodeString(value)
if err != nil {
return config.Settings{}, fmt.Errorf("decode model worker settings: %w", err)
}
var settings config.Settings
if err := json.Unmarshal(data, &settings); err != nil {
return config.Settings{}, fmt.Errorf("parse model worker settings: %w", err)
}
if err := settings.Validate(); err != nil {
return config.Settings{}, fmt.Errorf("validate model worker settings: %w", err)
}
return settings, nil
}
// RunModelWorker owns all Sherpa/VAD resources in the supervised child. It
// returns a process exit code so command entrypoints can call os.Exit.
func RunModelWorker(diagnostics io.Writer) int {
protocol := os.NewFile(uintptr(modelProtocolFD), "captioneer-model-protocol")
if protocol == nil {
fmt.Fprintln(diagnostics, "captioneer: model worker protocol is unavailable")
return 1
}
defer protocol.Close()
syscall.CloseOnExec(modelProtocolFD)
settings, err := decodeWorkerSettings(os.Getenv(modelSettingsEnvironment))
if err != nil {
fmt.Fprintf(diagnostics, "captioneer: %v\n", err)
return 1
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
reporter := &workerReporter{encoder: json.NewEncoder(protocol)}
hooks := directHooks{
ready: func() error { return reporter.send(workerMessage{Type: messageReady}) },
busy: func() error { return reporter.send(workerMessage{Type: messageBusy}) },
idle: func() error { return reporter.send(workerMessage{Type: messageIdle}) },
}
heartbeatInterval := time.Second
if settings.ModelTimeout > 0 && settings.ModelTimeout < 4*heartbeatInterval {
heartbeatInterval = settings.ModelTimeout / 4
if heartbeatInterval <= 0 {
heartbeatInterval = time.Millisecond
}
}
heartbeatContext, stopHeartbeat := context.WithCancel(ctx)
heartbeatDone := make(chan struct{})
go func() {
defer close(heartbeatDone)
ticker := time.NewTicker(heartbeatInterval)
defer ticker.Stop()
for {
select {
case <-heartbeatContext.Done():
return
case <-ticker.C:
if reporter.send(workerMessage{Type: messageHeartbeat}) != nil {
return
}
}
}
}()
err = runDirect(ctx, settings, protocolSink{reporter: reporter}, diagnostics, hooks)
stopHeartbeat()
<-heartbeatDone
if err == nil || errors.Is(err, context.Canceled) {
return 0
}
_ = reporter.send(workerMessage{Type: messageFailure, Error: err.Error()})
fmt.Fprintf(diagnostics, "captioneer: model worker: %v\n", err)
return 1
}
+2
View File
@@ -10,6 +10,7 @@ import (
"os/exec"
"strings"
"sync"
"syscall"
"tea.chunkbyte.com/kato/captioneer/src/audio"
)
@@ -41,6 +42,7 @@ func Packets(ctx context.Context, monitorSource string) (<-chan []float32, func(
"--latency-msec=50",
)
cmd.Stderr = os.Stderr
cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGKILL}
stdout, err := cmd.StdoutPipe()
if err != nil {
return nil, nil, err
+4
View File
@@ -20,6 +20,10 @@ import (
)
func main() {
if app.IsModelWorker() {
os.Exit(app.RunModelWorker(os.Stderr))
}
runtime.LockOSThread()
defer runtime.UnlockOSThread()
+4
View File
@@ -13,6 +13,10 @@ import (
)
func main() {
if app.IsModelWorker() {
os.Exit(app.RunModelWorker(os.Stderr))
}
settings := config.Parse()
if err := settings.Validate(); err != nil {
log.Fatal(err)
+17
View File
@@ -12,6 +12,8 @@ const (
DefaultModelsDir = "models"
DefaultOverlayFontFamily = "Sans"
MinimumOverlayFinalLifetime = 3 * time.Second
DefaultModelTimeout = 30 * time.Second
DefaultModelShutdownTimeout = 2 * time.Second
)
type Mode string
@@ -27,6 +29,9 @@ type Settings struct {
ModelsDir string
Threads int
PreviewThresholdDB float64
ModelAutoRestart bool
ModelTimeout time.Duration
ModelShutdownTimeout time.Duration
}
func Parse() Settings {
@@ -37,6 +42,9 @@ func Parse() Settings {
flag.StringVar(&settings.ModelsDir, "models-dir", DefaultModelsDir, "directory containing downloaded Sherpa models")
flag.IntVar(&settings.Threads, "threads", 2, "CPU threads used by recognition and VAD")
flag.Float64Var(&settings.PreviewThresholdDB, "preview-threshold-dbfs", -45, "RMS dBFS threshold used to begin VAD previews")
flag.BoolVar(&settings.ModelAutoRestart, "model-auto-restart", true, "restart the isolated model worker after a crash or timeout")
flag.DurationVar(&settings.ModelTimeout, "model-timeout", DefaultModelTimeout, "maximum model startup or processing time; zero disables hang detection")
flag.DurationVar(&settings.ModelShutdownTimeout, "model-shutdown-timeout", DefaultModelShutdownTimeout, "time allowed for model flush and cleanup before force-killing the worker")
flag.Parse()
settings.Mode = Mode(mode)
return settings
@@ -87,6 +95,9 @@ func ParseDesktop() DesktopSettings {
flags.StringVar(&desktop.ModelsDir, "models-dir", DefaultModelsDir, "directory containing downloaded Sherpa models")
flags.IntVar(&desktop.Threads, "threads", 2, "CPU threads used by recognition and VAD")
flags.Float64Var(&desktop.PreviewThresholdDB, "preview-threshold-dbfs", -45, "RMS dBFS threshold used to begin VAD previews")
flags.BoolVar(&desktop.ModelAutoRestart, "model-auto-restart", true, "restart the isolated model worker after a crash or timeout")
flags.DurationVar(&desktop.ModelTimeout, "model-timeout", DefaultModelTimeout, "maximum model startup or processing time; zero disables hang detection")
flags.DurationVar(&desktop.ModelShutdownTimeout, "model-shutdown-timeout", DefaultModelShutdownTimeout, "time allowed for model flush and cleanup before force-killing the worker")
flags.StringVar(&output, "output", "", "caption output: terminal, overlay, or both (required)")
flags.StringVar(&backend, "overlay-backend", "auto", "display backend: auto, wayland, or x11")
flags.Float64Var(&desktop.Overlay.Opacity, "overlay-opacity", 0.90, "caption background opacity from 0 to 1")
@@ -114,6 +125,12 @@ func (s Settings) Validate() error {
if s.Threads <= 0 {
return errors.New("--threads must be positive")
}
if s.ModelTimeout < 0 {
return errors.New("--model-timeout cannot be negative")
}
if s.ModelShutdownTimeout <= 0 {
return errors.New("--model-shutdown-timeout must be positive")
}
return nil
}
+15 -2
View File
@@ -6,7 +6,12 @@ import (
)
func TestSettingsValidate(t *testing.T) {
valid := Settings{Mode: ModeFixed, ChunkDuration: time.Second, Threads: 1}
valid := Settings{
Mode: ModeFixed,
ChunkDuration: time.Second,
Threads: 1,
ModelShutdownTimeout: DefaultModelShutdownTimeout,
}
if err := valid.Validate(); err != nil {
t.Fatalf("valid settings failed: %v", err)
}
@@ -17,7 +22,13 @@ func TestSettingsValidate(t *testing.T) {
func validDesktopSettings() DesktopSettings {
return DesktopSettings{
Settings: Settings{Mode: ModeVAD, ChunkDuration: time.Second, Threads: 2},
Settings: Settings{
Mode: ModeVAD,
ChunkDuration: time.Second,
Threads: 2,
ModelTimeout: DefaultModelTimeout,
ModelShutdownTimeout: DefaultModelShutdownTimeout,
},
Output: OutputOverlay,
Overlay: OverlaySettings{
Backend: BackendAuto, Opacity: 0.9, FontSize: 28, FontFamily: DefaultOverlayFontFamily, BottomMargin: 100,
@@ -46,6 +57,8 @@ func TestDesktopSettingsValidate(t *testing.T) {
{"empty monitor", func(s *DesktopSettings) { s.Overlay.Monitor = "" }},
{"negative timeout", func(s *DesktopSettings) { s.Overlay.FinalTimeout = -time.Second }},
{"short timeout", func(s *DesktopSettings) { s.Overlay.FinalTimeout = time.Second }},
{"negative model timeout", func(s *DesktopSettings) { s.ModelTimeout = -time.Second }},
{"zero model shutdown timeout", func(s *DesktopSettings) { s.ModelShutdownTimeout = 0 }},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
+5
View File
@@ -25,6 +25,11 @@ type Resources struct {
vad *sherpa.VoiceActivityDetector
}
func Validate(settings config.Settings) error {
_, err := validatePaths(settings.ModelsDir, settings.Mode)
return err
}
func New(settings config.Settings) (*Resources, error) {
paths, err := validatePaths(settings.ModelsDir, settings.Mode)
if err != nil {