diff --git a/README.md b/README.md index 3cfcb016..6048bd24 100644 --- a/README.md +++ b/README.md @@ -355,11 +355,17 @@ echo "setting set reading#mode=highlight" | socat - UNIX-CLIENT:/tmp/fenrirscree Set `remote.enabled = false` to disable remote control. Command and settings requests can also be enabled separately in the `[remote]` section. -`command notify TEXT` plays `notification.wav` when sounds are enabled, waits -for the sound to finish, then speaks the notification. It retains the message -for **Fenrir+N** and **Fenrir+Control+N**. History holds the latest 100 notifications for this reader -session, including messages received while speech is muted or temporarily -silenced. `command say TEXT` remains self-voicing and does not add history. +`command notify TEXT` plays `notification.wav` when sounds are enabled, at most +once every 30 seconds across readers for the same user. It waits for a played +cue to finish before speaking. Notifications during the cooldown are spoken +without another sound; skipped cues are not queued for later. Only the targeted reader speaks or plays +the cue. All running readers for that user share the latest 100 notifications +for **Fenrir+N** and **Fenrir+Control+N**, including messages received while +speech is muted or temporarily silenced. Other readers receive silent history +updates. The root console reader shares the active text-console user's history; +switching console users switches histories. A newly started reader loads the +current history silently. History stays in memory until the last reader for +that user exits. `command say TEXT` remains self-voicing and does not add history. Received times use the current time format in Presentation's Time and date settings. In the history list, use Up/Down to browse, Enter to repeat an entry, and Escape or the Close button to return to the application. Reopen the list diff --git a/docs/fenrir.1 b/docs/fenrir.1 index 793c2cc1..708ccc64 100644 --- a/docs/fenrir.1 +++ b/docs/fenrir.1 @@ -976,13 +976,17 @@ the default theme), waits for the audio output to drain, then speaks the message. Sound playback runs outside the input loop. Missing or failed sound output leaves speech available. -Speech mute or temporary silence keeps the sound and history, with no -spoken message. +The cue plays at most once every 30 seconds across readers for the same +user. +Notifications during this cooldown are spoken without another sound or +sound wait; skipped cues are not queued for later. +Speech mute or temporary silence keeps history and allows the sound +under the same cooldown, with no spoken message. Recalling a notification does not replay its sound. .PP \f[B]Fenrir+N\f[R] repeats the most recent notification. -\f[B]Fenrir+Control+N\f[R] opens a list of the latest 100 notifications -for this reader session, newest first. +\f[B]Fenrir+Control+N\f[R] opens a list of the latest 100 shared +notifications, newest first. Each message is followed by its received time, formatted using the current Presentation Time and date preference. For example, \f[CR]Hello world 01:38 PM\f[R] with 12\-hour time, or @@ -992,11 +996,32 @@ Use Up/Down to browse, Enter to repeat the selected entry, and Escape or the Close button to leave. Tab and Shift+Tab move between controls. The list is a snapshot; reopen it to include new arrivals. -History is cleared when the reader exits. Notifications received while muted or temporarily silenced are still remembered. Self\-voicing through \f[CR]command say TEXT\f[R] does not add notifications. +.PP +History is shared across running readers for the same account on this +machine. +Only the targeted reader speaks and plays an eligible cue; the other +readers receive silent history updates. +You can recall those messages from any reader. +The root console reader shares the owner of the active text console\(cqs +history. +Switching console users switches histories and closes the previous +user\(cqs notification dialog. +While a graphical VT is active, it keeps the last selected console +account. +A newly started reader loads existing history silently. +.PP +Notification text stays in memory; sharing needs no additional daemon or +init system service. +A surviving reader takes over when the coordinating reader exits. +History clears after the last reader for that account exits. +A crash can lose the most recent updates if other readers have not +received them yet. +All participating copies must run a build with shared notification +history. .SS Socket commands Unix socket remote control is enabled by default. The standard socket is \f[CR]/tmp/fenrirscreenreader\-daemon.sock\f[R], diff --git a/docs/user.md b/docs/user.md index fa13a730..8497edcb 100644 --- a/docs/user.md +++ b/docs/user.md @@ -652,20 +652,37 @@ are enabled and their volume is above zero, Fenrir plays the theme's `notification` cue (`notification.wav` in the default theme), waits for the audio output to drain, then speaks the message. Sound playback runs outside the input loop. Missing or failed sound output leaves speech available. -Speech mute or temporary silence keeps the sound and history, with no spoken -message. Recalling a notification does not replay its sound. +The cue plays at most once every 30 seconds across readers for the same user. +Notifications during this cooldown are spoken without another sound or sound +wait; skipped cues are not queued for later. Speech mute or temporary silence +keeps history and allows the sound under the same cooldown, with no spoken message. Recalling +a notification does not replay its sound. **Fenrir+N** repeats the most recent notification. **Fenrir+Control+N** opens -a list of the latest 100 notifications for this reader session, newest first. Each message +a list of the latest 100 shared notifications, newest first. Each message is followed by its received time, formatted using the current Presentation Time and date preference. For example, `Hello world 01:38 PM` with 12-hour time, or `Hello world 13:38` with 24-hour time. Use Up/Down to browse, Enter to repeat the selected entry, and Escape or the Close button to leave. Tab and Shift+Tab move between controls. The list is a -snapshot; reopen it to include new arrivals. History is cleared when the reader -exits. Notifications received while muted or temporarily silenced are still -remembered. Self-voicing through `command say TEXT` does not add notifications. +snapshot; reopen it to include new arrivals. Notifications received while muted +or temporarily silenced are still remembered. Self-voicing through +`command say TEXT` does not add notifications. + +History is shared across running readers for the same account on this machine. +Only the targeted reader speaks and plays an eligible cue; the other readers +receive silent history updates. You can recall those messages from any reader. +The root console reader shares the owner of the active text console's history. +Switching console users switches histories and closes the previous user's +notification dialog. While a graphical VT is active, it keeps the last selected +console account. A newly started reader loads existing history silently. + +Notification text stays in memory; sharing needs no additional daemon or init +system service. A surviving reader takes over when the coordinating reader +exits. History clears after the last reader for that account exits. A crash +can lose the most recent updates if other readers have not received them yet. +All participating copies must run a build with shared notification history. ### Socket commands diff --git a/docs/user.txt b/docs/user.txt index 9be17578..597c857c 100644 --- a/docs/user.txt +++ b/docs/user.txt @@ -704,23 +704,40 @@ sounds are enabled and their volume is above zero, Fenrir plays the theme’s notification cue (notification.wav in the default theme), waits for the audio output to drain, then speaks the message. Sound playback runs outside the input loop. Missing or failed sound output leaves -speech available. Speech mute or temporary silence keeps the sound and -history, with no spoken message. Recalling a notification does not -replay its sound. +speech available. The cue plays at most once every 30 seconds across +readers for the same user. Notifications during this cooldown are spoken +without another sound or sound wait; skipped cues are not queued for +later. Speech mute or temporary silence keeps history and allows the +sound under the same cooldown, with no spoken message. Recalling a +notification does not replay its sound. Fenrir+N repeats the most recent notification. Fenrir+Control+N opens a -list of the latest 100 notifications for this reader session, newest -first. Each message is followed by its received time, formatted using -the current Presentation Time and date preference. For example, -Hello world 01:38 PM with 12-hour time, or Hello world 13:38 with -24-hour time. +list of the latest 100 shared notifications, newest first. Each message +is followed by its received time, formatted using the current +Presentation Time and date preference. For example, Hello world 01:38 PM +with 12-hour time, or Hello world 13:38 with 24-hour time. Use Up/Down to browse, Enter to repeat the selected entry, and Escape or the Close button to leave. Tab and Shift+Tab move between controls. The -list is a snapshot; reopen it to include new arrivals. History is -cleared when the reader exits. Notifications received while muted or -temporarily silenced are still remembered. Self-voicing through -command say TEXT does not add notifications. +list is a snapshot; reopen it to include new arrivals. Notifications +received while muted or temporarily silenced are still remembered. +Self-voicing through command say TEXT does not add notifications. + +History is shared across running readers for the same account on this +machine. Only the targeted reader speaks and plays an eligible cue; the +other readers receive silent history updates. You can recall those +messages from any reader. The root console reader shares the owner of +the active text console’s history. Switching console users switches +histories and closes the previous user’s notification dialog. While a +graphical VT is active, it keeps the last selected console account. A +newly started reader loads existing history silently. + +Notification text stays in memory; sharing needs no additional daemon or +init system service. A surviving reader takes over when the coordinating +reader exits. History clears after the last reader for that account +exits. A crash can lose the most recent updates if other readers have +not received them yet. All participating copies must run a build with +shared notification history. Socket commands diff --git a/internal/app/app.go b/internal/app/app.go index 947d8437..24a9e2d6 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -90,4 +90,7 @@ type RunOptions struct { Remote *control.Server TerminalMode bool Spellcheck *spellcheck.Service + + SharedNotifications bool + NotificationNamespace string // Private IPC namespace for native fixtures. } diff --git a/internal/app/application.go b/internal/app/application.go index cfb34992..3bbac789 100644 --- a/internal/app/application.go +++ b/internal/app/application.go @@ -20,10 +20,12 @@ package app import ( "context" "log/slog" + "sync" "time" "git.stormux.org/storm/fenrir/internal/config" "git.stormux.org/storm/fenrir/internal/i18n" + "git.stormux.org/storm/fenrir/internal/notifications" "git.stormux.org/storm/fenrir/internal/preferences" "git.stormux.org/storm/fenrir/internal/presentation" "git.stormux.org/storm/fenrir/internal/prompt" @@ -58,6 +60,10 @@ type application struct { speechChoices chan preferences.Discovery prepareSpeech func(config.Speech) (speech.Transport, error) + notificationSound notificationSoundGate + notificationHistory *notifications.Client + notificationClosers sync.WaitGroup + muted, temporarySilence bool quitting, quitAfterSave bool diff --git a/internal/app/event_loop.go b/internal/app/event_loop.go index 0954f25c..3941883b 100644 --- a/internal/app/event_loop.go +++ b/internal/app/event_loop.go @@ -66,6 +66,12 @@ func run(ctx context.Context, source Source, keyboards Keyboards, voices Speaker return err } defer func() { a.ui.close(); a.ui.testWorkers.Wait() }() + defer func() { + if a.notificationHistory != nil { + a.notificationHistory.Close() + } + a.notificationClosers.Wait() + }() defer func() { for _, selected := range a.overrides { selected.adapter.Close() @@ -122,7 +128,13 @@ func (a *application) loop(ticks <-chan time.Time) error { if a.quitting { return nil } + var notificationUpdates <-chan struct{} + if a.notificationHistory != nil { + notificationUpdates = a.notificationHistory.Updates() + } select { + case <-notificationUpdates: + a.refresh_notification_history() case discovery := <-a.speechChoices: a.quick.discovery = discovery case loaded := <-a.overrideLoads: diff --git a/internal/app/notifications.go b/internal/app/notifications.go index cc06a638..ca819b7a 100644 --- a/internal/app/notifications.go +++ b/internal/app/notifications.go @@ -5,54 +5,162 @@ package app import ( "context" + "fmt" "log/slog" + "os" "strings" + "sync" + "syscall" "time" "git.stormux.org/storm/fenrir/internal/datetime" "git.stormux.org/storm/fenrir/internal/diagnostic" "git.stormux.org/storm/fenrir/internal/i18n" "git.stormux.org/storm/fenrir/internal/interaction" + "git.stormux.org/storm/fenrir/internal/notifications" "git.stormux.org/storm/fenrir/internal/presentation" ) -const notificationLimit = 100 +const notificationLimit = notifications.Limit +const notificationSoundInterval = notifications.SoundInterval + +// The speech worker and muted notifications on the event loop share this gate. +// Check at playback admission so queued messages cannot accumulate sound waits. +type notificationSoundGate struct { + mu sync.Mutex + lastPlay time.Time +} + +func (g *notificationSoundGate) allow(now time.Time) bool { + g.mu.Lock() + defer g.mu.Unlock() + if !g.lastPlay.IsZero() && now.Sub(g.lastPlay) < notificationSoundInterval { + return false + } + g.lastPlay = now + return true +} type notification struct { text string received time.Time } -// Notifications belong to this reader session, independently of self-voicing -// and the last speech echo. Keep them even when speech is silenced or muted. +// Share only the active console account's history with its terminal readers. +// On graphical VTs keep the previous account until a text console is active. +func (a *application) share_notification_history(terminal int, active bool) { + if !a.options.SharedNotifications { + return + } + uid := os.Geteuid() + if uid == 0 && !a.options.TerminalMode { + if !active && a.notificationHistory != nil { + return + } + info, err := os.Stat(fmt.Sprintf("/dev/tty%d", terminal)) + if err != nil { + a.clear_notification_user() + return + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok { + a.clear_notification_user() + return + } + uid = int(stat.Uid) + } + a.select_notification_user(uid) +} + +func (a *application) select_notification_user(uid int) { + if previous := a.notificationHistory; previous != nil && previous.UID == uid { + return + } + a.clear_notification_user() + a.notificationHistory = notifications.New(a.ctx, uid, a.options.NotificationNamespace) +} + +func (a *application) clear_notification_user() { + if previous := a.notificationHistory; previous != nil { + a.voice.Interrupt() + a.ui.close() + a.notificationClosers.Go(previous.Close) + } + a.notifications = nil + a.notificationHistory = nil +} + +// Snapshot updates are deliberately silent: never call notify, present, or the +// sound service while incorporating another reader's history. +func (a *application) refresh_notification_history() { + if a.notificationHistory == nil { + return + } + state := a.notificationHistory.Snapshot() + a.notifications = make([]notification, 0, len(state.Entries)) + for _, entry := range state.Entries { + a.notifications = append(a.notifications, notification{text: entry.Text, received: entry.Received.In(time.Local)}) + } +} + +// Only a targeted notification calls this method. Shared snapshots update +// history separately, without automatic speech or sound. func (a *application) notify(text string, received time.Time) { text = strings.TrimSpace(text) if text == "" { return } - if len(a.notifications) == notificationLimit { - copy(a.notifications, a.notifications[1:]) - a.notifications = a.notifications[:notificationLimit-1] + a.share_notification_history(a.lastTerminal, a.lastActive) + if a.notificationHistory != nil { + a.notificationHistory.Publish(text, received) + a.refresh_notification_history() + } else { + if len(a.notifications) == notificationLimit { + copy(a.notifications, a.notifications[1:]) + a.notifications = a.notifications[:notificationLimit-1] + } + a.notifications = append(a.notifications, notification{text: text, received: received}) } - a.notifications = append(a.notifications, notification{text: text, received: received}) diagnostic.Record("notification received", "length", len(text), "retained", len(a.notifications)) if a.settings.live.Presentation.Sound.Enabled && a.settings.live.Presentation.Sound.Volume > 0 && a.cues != nil { + cues := a.cues + history := a.notificationHistory if a.temporarySilence || a.muted { - a.play(presentation.Notification) + if history != nil { + history.Sound(func(ctx context.Context) { play_notification(ctx, cues) }) + } else if a.notificationSound.allow(time.Now()) { + a.play(presentation.Notification) + } return } - cues := a.cues + gate := &a.notificationSound a.voice.Say_after(text, func(ctx context.Context) { - if err := cues.Play_wait(ctx, presentation.Notification); err != nil && ctx.Err() == nil { - slog.Warn("notification sound incomplete", "error", err, "at", time.Now()) + if ctx.Err() != nil { + return } + if history != nil { + if !history.Claim_sound(ctx) { + return + } + } else if !gate.allow(time.Now()) { + return + } + play_notification(ctx, cues) }) return } a.present(text, "") } +func play_notification(ctx context.Context, cues SoundPlayer) { + if err := cues.Play_wait(ctx, presentation.Notification); err != nil && ctx.Err() == nil { + slog.Warn("notification sound incomplete", "error", err, "at", time.Now()) + } +} + func (a *application) last_notification() { + a.share_notification_history(a.lastTerminal, a.lastActive) + a.refresh_notification_history() if len(a.notifications) == 0 { a.present(i18n.T("No notifications"), "") return @@ -61,6 +169,8 @@ func (a *application) last_notification() { } func (a *application) open_notifications() { + a.share_notification_history(a.lastTerminal, a.lastActive) + a.refresh_notification_history() if len(a.notifications) == 0 { a.present(i18n.T("No notifications"), "") return diff --git a/internal/app/notifications_test.go b/internal/app/notifications_test.go index 552bea6e..3ab0b0be 100644 --- a/internal/app/notifications_test.go +++ b/internal/app/notifications_test.go @@ -13,6 +13,7 @@ import ( "path/filepath" "reflect" "strings" + "sync" "testing" "time" @@ -23,6 +24,125 @@ import ( "git.stormux.org/storm/fenrir/internal/presentation" ) +func TestNotificationSoundCooldownBoundaryAndConcurrentAdmission(t *testing.T) { + var gate notificationSoundGate + now := time.Now() + var workers sync.WaitGroup + accepted := make(chan bool, 32) + for range cap(accepted) { + workers.Go(func() { accepted <- gate.allow(now) }) + } + workers.Wait() + close(accepted) + count := 0 + for allowed := range accepted { + if allowed { + count++ + } + } + if count != 1 { + t.Fatal("concurrent notifications admitted multiple cues", count) + } + for _, trial := range []struct { + after time.Duration + want bool + }{ + {time.Second, false}, + {30*time.Second - time.Nanosecond, false}, + {30 * time.Second, true}, + {31 * time.Second, false}, + {60 * time.Second, true}, + } { + if got := gate.allow(now.Add(trial.after)); got != trial.want { + t.Fatalf("after %v: allowed=%v, want %v", trial.after, got, trial.want) + } + } +} + +func TestNotificationBurstLimitsCuesWithoutLosingSpeechOrHistory(t *testing.T) { + for _, mode := range []string{"enabled", "muted", "silenced"} { + t.Run(mode, func(t *testing.T) { + a, voice := persistence_application(t, true) + voice.messages = nil + sounds := &fakeSounds{} + a.cues = sounds + a.muted, voice.muted = mode == "muted", mode == "muted" + a.temporarySilence = mode == "silenced" + var want []string + for index := range 10 { + text := fmt.Sprintf("Notification %d", index) + want = append(want, text) + a.notify(text, time.Now()) + if len(sounds.cues) != 1 || len(a.notifications) != index+1 || a.notifications[index].text != text { + t.Fatal("burst cue or retained message", sounds.cues, a.notifications) + } + } + if mode != "enabled" { + want = nil + } + if !reflect.DeepEqual(voice.messages, want) { + t.Fatal("burst changed speech policy", voice.messages) + } + // A mute transition shares the cooldown with ordinary notifications. + a.muted, voice.muted, a.temporarySilence = false, false, false + a.notify("Still cooling down", time.Now()) + if len(sounds.cues) != 1 || voice.messages[len(voice.messages)-1] != "Still cooling down" { + t.Fatal("unmuting reset the cooldown", sounds.cues, voice.messages) + } + a.notificationSound.mu.Lock() + a.notificationSound.lastPlay = time.Now().Add(-notificationSoundInterval) + a.notificationSound.mu.Unlock() + a.notify("After cooldown", time.Now()) + if len(sounds.cues) != 2 || voice.messages[len(voice.messages)-1] != "After cooldown" || len(a.notifications) != 12 { + t.Fatal("cue did not resume after cooldown", sounds.cues, voice.messages, a.notifications) + } + }) + } +} + +type queuedNotificationVoice struct { + fakeVoice + requests []struct { + text string + before func(context.Context) + } +} + +func (v *queuedNotificationVoice) Say_after(text string, before func(context.Context)) { + v.requests = append(v.requests, struct { + text string + before func(context.Context) + }{text, before}) +} + +func TestQueuedNotificationsCheckCooldownAtPlaybackAndCancellation(t *testing.T) { + a, _ := persistence_application(t, true) + voice := &queuedNotificationVoice{} + a.voice.Speaker = voice + sounds := &fakeSounds{} + a.cues = sounds + a.notify("Canceled", time.Now()) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + voice.requests[0].before(ctx) + if !a.notificationSound.lastPlay.IsZero() || len(sounds.cues) != 0 { + t.Fatal("canceled cue started the cooldown") + } + a.notify("First queued", time.Now()) + // More than 30 seconds can pass while notifications wait in the speech queue. + a.notify("Second queued", time.Now().Add(time.Minute)) + if !a.notificationSound.lastPlay.IsZero() || len(sounds.cues) != 0 { + t.Fatal("cue cooldown started before playback") + } + for _, request := range voice.requests[1:] { + request.before(context.Background()) + voice.Say(request.text) + } + if !reflect.DeepEqual(sounds.cues, []presentation.Cue{presentation.Notification}) || !reflect.DeepEqual(voice.messages, []string{"First queued", "Second queued"}) { + t.Fatal("queued notifications accumulated sound waits or lost speech", sounds.cues, voice.messages) + } +} + func TestNotificationCuePrecedesSpeechAndOnlyPlaysForNewNotifications(t *testing.T) { for _, mode := range []string{"enabled", "disabled", "zero volume", "missing cue", "unavailable", "muted", "silenced"} { t.Run(mode, func(t *testing.T) { diff --git a/internal/app/screen_updates.go b/internal/app/screen_updates.go index f882463f..4c3e9c1e 100644 --- a/internal/app/screen_updates.go +++ b/internal/app/screen_updates.go @@ -34,6 +34,7 @@ func (a *application) read_screen() error { if err != nil { return err } + a.share_notification_history(snapshot.Terminal, snapshot.Active) if a.lastTerminal == 0 { if snapshot.Active { a.present(i18n.T("Start Fenrir"), presentation.Ready) diff --git a/internal/app/shared_notifications_test.go b/internal/app/shared_notifications_test.go new file mode 100644 index 00000000..0ac0e467 --- /dev/null +++ b/internal/app/shared_notifications_test.go @@ -0,0 +1,127 @@ +// Copyright (c) 2026 Storm Dragon +// SPDX-License-Identifier: GPL-3.0-or-later + +package app + +import ( + "fmt" + "os" + "reflect" + "testing" + "time" +) + +func shared_notification_application(t *testing.T, namespace string) (*application, *fakeVoice, *fakeSounds) { + t.Helper() + a, voice := persistence_application(t, true) + voice.messages = nil + sounds := &fakeSounds{} + a.cues = sounds + a.options.SharedNotifications = true + a.options.NotificationNamespace = namespace + a.share_notification_history(1, true) + t.Cleanup(func() { + a.notificationHistory.Close() + a.notificationClosers.Wait() + }) + return a, voice, sounds +} + +func wait_shared_notifications(t *testing.T, a *application, count int) { + t.Helper() + deadline := time.NewTimer(3 * time.Second) + defer deadline.Stop() + for { + a.refresh_notification_history() + if len(a.notifications) == count { + return + } + select { + case <-a.notificationHistory.Updates(): + case <-deadline.C: + t.Fatalf("notification history has %d entries, want %d", len(a.notifications), count) + } + } +} + +func TestSharedNotificationHistoryNeverSpeaksOrPlaysOnOtherReaders(t *testing.T) { + namespace := fmt.Sprintf(".app.%d.%d", os.Getpid(), time.Now().UnixNano()) + var readers []*application + var voices []*fakeVoice + var sounds []*fakeSounds + for range 4 { + a, voice, cues := shared_notification_application(t, namespace) + readers, voices, sounds = append(readers, a), append(voices, voice), append(sounds, cues) + } + if response := runtime_request(readers[0], "command notify First target"); response != "" { + t.Fatal(response) + } + for index, a := range readers { + wait_shared_notifications(t, a, 1) + wantSpeech, wantCues := 0, 0 + if index == 0 { + wantSpeech, wantCues = 1, 1 + } + if len(voices[index].messages) != wantSpeech || len(sounds[index].cues) != wantCues { + t.Fatal("silent history update triggered speech or sound", index, voices[index].messages, sounds[index].cues) + } + } + runtime_request(readers[1], "command notify Second target") + for index, a := range readers { + wait_shared_notifications(t, a, 2) + if !reflect.DeepEqual(a.notifications, readers[0].notifications) { + t.Fatal("readers disagree about history", index) + } + } + if !reflect.DeepEqual(voices[0].messages, []string{"First target"}) || !reflect.DeepEqual(voices[1].messages, []string{"Second target"}) { + t.Fatal("targeted speech was duplicated", voices[0].messages, voices[1].messages) + } + for index := 1; index < len(readers); index++ { + if len(sounds[index].cues) != 0 { + t.Fatal("target switch reset the shared sound cooldown", index, sounds[index].cues) + } + } + // History recall is an explicit user action and still belongs to that reader. + readers[2].last_notification() + if !reflect.DeepEqual(voices[2].messages, []string{"Second target"}) || len(sounds[2].cues) != 0 { + t.Fatal("history recall was missing or replayed the cue") + } + readers[2].muted, voices[2].muted = true, true + runtime_request(readers[2], "command notify Muted target") + readers[3].temporarySilence = true + runtime_request(readers[3], "command notify Silenced target") + for index, a := range readers { + wait_shared_notifications(t, a, 4) + if index >= 2 && len(sounds[index].cues) != 0 { + t.Fatal("muted or silenced target replayed the cue") + } + } + joining, voice, cues := shared_notification_application(t, namespace) + wait_shared_notifications(t, joining, 4) + if len(voice.messages) != 0 || len(cues.cues) != 0 || !reflect.DeepEqual(joining.notifications, readers[0].notifications) { + t.Fatal("joining reader announced existing history or loaded different messages") + } +} + +func TestNotificationUserSwitchClearsHistoryAndClosesOldDialog(t *testing.T) { + namespace := fmt.Sprintf(".app.%d.%d", os.Getpid(), time.Now().UnixNano()) + a, voice, _ := shared_notification_application(t, namespace) + peer, _, _ := shared_notification_application(t, namespace) + a.notify("Original account", time.Now()) + wait_shared_notifications(t, peer, 1) + a.open_notifications() + if !a.ui.active { + t.Fatal("notification dialog did not open") + } + a.select_notification_user(os.Geteuid() + 1) + voice.messages = nil + a.refresh_notification_history() + if len(a.notifications) != 0 || a.ui.active { + t.Fatal("switching accounts exposed the previous history or dialog") + } + a.select_notification_user(os.Geteuid()) + wait_shared_notifications(t, a, 1) + if len(voice.messages) != 0 { + t.Fatal("restoring an account spoke its history", voice.messages) + } +} diff --git a/internal/app/startup.go b/internal/app/startup.go index 40cba742..74758ea6 100644 --- a/internal/app/startup.go +++ b/internal/app/startup.go @@ -196,7 +196,7 @@ func start(ctx context.Context, settings config.Settings, terminalMode bool, con } transfers := clipboard.New_worker(ctx, paste) defer transfers.Close() - result = Run(ctx, source, keyboards, voices, RunOptions{Ready: ready, Spellcheck: spellChecker, Bindings: bindings.Bindings, Settings: settings, EffectiveSpeech: effectiveSpeech, Punctuation: punctuationProcessor, Sounds: cues, ConfigPath: configPath, Clipboard: transfers, ClipboardHistory: historyStore, SettingsChoices: preferences.Discover_choices, Scripts: scriptCommands, ScriptRunner: scriptRunner, Remote: remote, TerminalMode: terminalMode}) + result = Run(ctx, source, keyboards, voices, RunOptions{Ready: ready, Spellcheck: spellChecker, Bindings: bindings.Bindings, Settings: settings, EffectiveSpeech: effectiveSpeech, Punctuation: punctuationProcessor, Sounds: cues, ConfigPath: configPath, Clipboard: transfers, ClipboardHistory: historyStore, SettingsChoices: preferences.Discover_choices, Scripts: scriptCommands, ScriptRunner: scriptRunner, Remote: remote, TerminalMode: terminalMode, SharedNotifications: true}) if result == nil && player != nil { if err := player.Finish(context.Background()); err != nil { slog.Warn("shutdown sound incomplete", "error", err, "at", time.Now()) diff --git a/internal/notifications/shared_linux.go b/internal/notifications/shared_linux.go new file mode 100644 index 00000000..84e3aee7 --- /dev/null +++ b/internal/notifications/shared_linux.go @@ -0,0 +1,418 @@ +// Copyright (c) 2026 Storm Dragon +// SPDX-License-Identifier: GPL-3.0-or-later + +// Package notifications shares silent history and cue admission between readers. +// The first reader hosts an in-memory coordinator on a per-user abstract Unix +// socket. No notification text is written to disk and no service manager is used. +package notifications + +import ( + "context" + "crypto/rand" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net" + "os" + "slices" + "strings" + "sync" + "syscall" + "time" + "unicode/utf8" +) + +const Limit = 100 +const SoundInterval = 30 * time.Second +const exchangeTimeout = 400 * time.Millisecond + +type Entry struct { + ID string + Text string + Received time.Time +} + +type Snapshot struct { + Entries []Entry + LastSound time.Time +} + +type request struct { + Version int + Seed Snapshot + Claim bool + Member string + Leave bool +} + +type response struct { + Version int + State Snapshot + Allowed bool +} + +type Client struct { + UID int + id string + address string + ctx context.Context + cancel context.CancelFunc + mu sync.Mutex + state Snapshot + brokerMu sync.Mutex + broker *broker + wake chan struct{} + updates chan struct{} + sounds chan func(context.Context) + done chan struct{} + closeOnce sync.Once +} + +// New starts only private IPC work. Namespace isolates native tests from live +// readers; production uses the empty namespace and the kernel account ID. +func New(ctx context.Context, uid int, namespace string) *Client { + ctx, cancel := context.WithCancel(ctx) + c := &Client{UID: uid, id: fmt.Sprintf("%x", random_id()), address: fmt.Sprintf("@fenrir-notifications.%d%s", uid, namespace), ctx: ctx, cancel: cancel, + wake: make(chan struct{}, 1), updates: make(chan struct{}, 1), sounds: make(chan func(context.Context), 1), done: make(chan struct{})} + go c.run() + return c +} + +func (c *Client) Updates() <-chan struct{} { return c.updates } + +func (c *Client) Snapshot() Snapshot { + c.mu.Lock() + defer c.mu.Unlock() + return Snapshot{Entries: slices.Clone(c.state.Entries), LastSound: c.state.LastSound} +} + +func (c *Client) Publish(text string, received time.Time) { + text = strings.TrimSpace(text) + if text == "" { + return + } + c.adopt(Snapshot{Entries: []Entry{{ID: fmt.Sprintf("%x", random_id()), Text: text, Received: received.UTC()}}}) + signal(c.wake) +} + +func random_id() [16]byte { + var id [16]byte + // crypto/rand.Read cannot return an error with the Go version we require. + _, _ = rand.Read(id[:]) + return id +} + +func signal(channel chan struct{}) { + select { + case channel <- struct{}{}: + default: + } +} + +func (c *Client) adopt(state Snapshot) { + c.mu.Lock() + before := c.state + c.state = merge(c.state, state) + changed := !slices.Equal(before.Entries, c.state.Entries) + c.mu.Unlock() + if changed { + signal(c.updates) + } +} + +// Claim_sound fails closed if coordination is unavailable. Notification speech +// never depends on its success, and history synchronization cannot call speech. +func (c *Client) Claim_sound(ctx context.Context) bool { + if last := c.Snapshot().LastSound; !last.IsZero() && time.Since(last) < SoundInterval { + return false + } + result, err := c.exchange(ctx, true) + return err == nil && result.Allowed +} + +// Sound is for muted notifications: coalesce pending cue requests without +// running network or sound waits on the reader's input loop. +func (c *Client) Sound(play func(context.Context)) { + select { + case c.sounds <- play: + default: + } +} + +func (c *Client) run() { + defer close(c.done) + tick := time.NewTicker(200 * time.Millisecond) + defer tick.Stop() + var previousError string + for { + _, err := c.exchange(c.ctx, false) + if err != nil && c.ctx.Err() == nil && err.Error() != previousError { + slog.Warn("notification history synchronization failed", "error", err, "at", time.Now()) + previousError = err.Error() + } else if err == nil { + previousError = "" + } + select { + case <-c.ctx.Done(): + return + case <-c.wake: + case <-tick.C: + case play := <-c.sounds: + if c.Claim_sound(c.ctx) { + play(c.ctx) + } + } + } +} + +func (c *Client) Close() { + c.closeOnce.Do(func() { + c.cancel() + <-c.done + // Flush the local outbox even when the reader's parent was canceled. + _, _ = c.transact(context.Background(), false, true) + c.brokerMu.Lock() + if c.broker != nil { + c.broker.close() + } + c.brokerMu.Unlock() + }) +} + +func (c *Client) exchange(ctx context.Context, claim bool) (response, error) { + return c.transact(ctx, claim, false) +} + +func (c *Client) transact(ctx context.Context, claim, leaving bool) (response, error) { + var result response + ctx, cancel := context.WithTimeout(ctx, exchangeTimeout) + defer cancel() + if !leaving { + stop := context.AfterFunc(c.ctx, cancel) + defer stop() + } + if c.UID < 0 || os.Geteuid() != 0 && os.Geteuid() != c.UID { + return result, fmt.Errorf("notification account is not permitted") + } + if err := ctx.Err(); err != nil { + return result, err + } + dialer := net.Dialer{} + connection, err := dialer.DialContext(ctx, "unix", c.address) + if errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ENOENT) { + if !leaving { + if err := c.host(); err != nil && !errors.Is(err, syscall.EADDRINUSE) { + return result, err + } + connection, err = dialer.DialContext(ctx, "unix", c.address) + } + } + if err != nil { + return result, err + } + defer connection.Close() + stopConnection := context.AfterFunc(ctx, func() { connection.Close() }) + defer stopConnection() + uid, err := peer_uid(connection.(*net.UnixConn)) + if err != nil || uid != c.UID && uid != 0 { + return result, fmt.Errorf("notification coordinator belongs to another account") + } + deadline, _ := ctx.Deadline() + _ = connection.SetDeadline(deadline) + if err := json.NewEncoder(connection).Encode(request{Version: 1, Seed: c.Snapshot(), Claim: claim, Member: c.id, Leave: leaving}); err != nil { + return result, err + } + if err := json.NewDecoder(io.LimitReader(connection, 8<<20)).Decode(&result); err != nil { + return result, err + } + if result.Version != 1 || !valid(result.State) { + return response{}, fmt.Errorf("invalid notification snapshot") + } + c.adopt(result.State) + return result, nil +} + +func (c *Client) host() error { + c.brokerMu.Lock() + defer c.brokerMu.Unlock() + if err := c.ctx.Err(); err != nil { + return err + } + listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: c.address, Net: "unix"}) + if err != nil { + return err + } + c.broker = &broker{listener: listener, uid: c.UID, state: c.Snapshot(), done: make(chan struct{}), members: make(map[string]acknowledgement), changed: make(chan struct{}, 1)} + go c.broker.serve() + return nil +} + +type acknowledgement struct { + state Snapshot + seen time.Time +} + +type broker struct { + listener *net.UnixListener + uid int + mu sync.Mutex + state Snapshot + workers sync.WaitGroup + done chan struct{} + members map[string]acknowledgement + changed chan struct{} +} + +func (b *broker) serve() { + defer close(b.done) + // Bound resource use independently of the reader's ordinary remote sockets. + clients := make(chan struct{}, 16) + for { + connection, err := b.listener.AcceptUnix() + if err != nil { + return + } + select { + case clients <- struct{}{}: + b.workers.Go(func() { + defer func() { connection.Close(); <-clients }() + b.handle(connection) + }) + default: + connection.Close() + } + } +} + +func (b *broker) handle(connection *net.UnixConn) { + uid, err := peer_uid(connection) + if err != nil || uid != 0 && uid != b.uid { + return + } + _ = connection.SetDeadline(time.Now().Add(exchangeTimeout)) + var message request + if json.NewDecoder(io.LimitReader(connection, 8<<20)).Decode(&message) != nil || message.Version != 1 || len(message.Member) != 32 || !valid(message.Seed) { + return + } + b.mu.Lock() + now := time.Now() + for id, member := range b.members { + if now.Sub(member.seen) > 5*time.Second { + delete(b.members, id) + } + } + // A crashed reader cannot deregister. Bound retained acknowledgement state + // as well as concurrent connections; history itself is always capped at 100. + if _, exists := b.members[message.Member]; !message.Leave && !exists && len(b.members) >= 128 { + b.mu.Unlock() + return + } + b.state = merge(b.state, message.Seed) + if message.Leave { + delete(b.members, message.Member) + } else { + b.members[message.Member] = acknowledgement{state: message.Seed, seen: now} + } + allowed := false + if message.Claim && (b.state.LastSound.IsZero() || now.Sub(b.state.LastSound) >= SoundInterval) { + b.state.LastSound, allowed = now, true + } + result := response{Version: 1, State: b.state, Allowed: allowed} + b.mu.Unlock() + signal(b.changed) + _ = json.NewEncoder(connection).Encode(result) +} + +func (b *broker) close() { + // Give surviving replicas time to acknowledge the final snapshot before + // releasing the coordinator's socket. This runs after releasing keyboards. + deadline := time.NewTimer(3 * exchangeTimeout) + defer deadline.Stop() +replicate: + for { + b.mu.Lock() + replicated := true + for _, member := range b.members { + state := member.state + merged := merge(state, b.state) + if !slices.Equal(merged.Entries, state.Entries) || merged.LastSound.After(state.LastSound) { + replicated = false + break + } + } + b.mu.Unlock() + if replicated { + break + } + select { + case <-b.changed: + case <-deadline.C: + break replicate + } + } + b.listener.Close() + <-b.done + b.workers.Wait() +} + +func peer_uid(connection *net.UnixConn) (int, error) { + raw, err := connection.SyscallConn() + if err != nil { + return -1, err + } + var credentials *syscall.Ucred + var credentialError error + err = raw.Control(func(fd uintptr) { + credentials, credentialError = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED) + }) + if err != nil { + return -1, err + } + if credentialError != nil { + return -1, credentialError + } + return int(credentials.Uid), nil +} + +func valid(state Snapshot) bool { + if len(state.Entries) > Limit || state.LastSound.After(time.Now().Add(time.Second)) { + return false + } + seen := make(map[string]bool, len(state.Entries)) + for _, entry := range state.Entries { + if len(entry.ID) != 32 || seen[entry.ID] || strings.TrimSpace(entry.Text) == "" || len(entry.Text) > 8129 || !utf8.ValidString(entry.Text) || entry.Received.IsZero() { + return false + } + seen[entry.ID] = true + } + return true +} + +func merge(left, right Snapshot) Snapshot { + entries := make(map[string]Entry, len(left.Entries)+len(right.Entries)) + for _, entry := range right.Entries { + entries[entry.ID] = entry + } + for _, entry := range left.Entries { + entries[entry.ID] = entry + } + result := Snapshot{LastSound: left.LastSound} + if right.LastSound.After(result.LastSound) { + result.LastSound = right.LastSound + } + for _, entry := range entries { + result.Entries = append(result.Entries, entry) + } + slices.SortFunc(result.Entries, func(a, b Entry) int { + if order := a.Received.Compare(b.Received); order != 0 { + return order + } + return strings.Compare(a.ID, b.ID) + }) + if len(result.Entries) > Limit { + result.Entries = result.Entries[len(result.Entries)-Limit:] + } + return result +} diff --git a/internal/notifications/shared_test.go b/internal/notifications/shared_test.go new file mode 100644 index 00000000..34079cc4 --- /dev/null +++ b/internal/notifications/shared_test.go @@ -0,0 +1,374 @@ +// Copyright (c) 2026 Storm Dragon +// SPDX-License-Identifier: GPL-3.0-or-later + +package notifications + +import ( + "context" + "encoding/json" + "fmt" + "net" + "os" + "os/exec" + "reflect" + "strings" + "sync" + "testing" + "time" +) + +func test_client(t *testing.T, namespace string) *Client { + t.Helper() + c := New(context.Background(), os.Geteuid(), namespace) + t.Cleanup(c.Close) + return c +} + +func wait_history(t *testing.T, c *Client, want int) Snapshot { + t.Helper() + deadline := time.NewTimer(5 * time.Second) + defer deadline.Stop() + for { + if state := c.Snapshot(); len(state.Entries) == want { + return state + } + select { + case <-c.Updates(): + case <-deadline.C: + t.Fatalf("history has %d entries, want %d", len(c.Snapshot().Entries), want) + } + } +} + +func TestSharedHistoryConcurrentPublishingBoundAndSingleCue(t *testing.T) { + namespace := fmt.Sprintf(".%x", random_id()) + var clients []*Client + for range 4 { + clients = append(clients, test_client(t, namespace)) + } + var workers sync.WaitGroup + for index, client := range clients { + workers.Go(func() { + for count := range 30 { + client.Publish(fmt.Sprintf("Reader %d message %d", index, count), time.Now()) + } + }) + } + workers.Wait() + for _, client := range clients { + wait_history(t, client, Limit) + } + // Reconcile after reaching the limit: all replicas must agree on the same + // newest 100 IDs rather than just having the same length. + for range 3 { + for _, client := range clients { + if _, err := client.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + } + } + want := clients[0].Snapshot().Entries + for _, client := range clients[1:] { + if !reflect.DeepEqual(client.Snapshot().Entries, want) { + t.Fatal("readers disagree about bounded history") + } + } + accepted := make(chan bool, len(clients)) + for _, client := range clients { + workers.Go(func() { accepted <- client.Claim_sound(context.Background()) }) + } + workers.Wait() + close(accepted) + count := 0 + for allowed := range accepted { + if allowed { + count++ + } + } + if count != 1 { + t.Fatal("multiple readers admitted the notification cue", count) + } + for _, client := range clients { + if client.Claim_sound(context.Background()) { + t.Fatal("a repeated cue escaped the shared cooldown") + } + } +} + +func TestCoordinatorTakeoverAndHistoryLifetime(t *testing.T) { + namespace := fmt.Sprintf(".%x", random_id()) + owner := test_client(t, namespace) + owner.Publish("Before takeover", time.Now()) + if !owner.Claim_sound(context.Background()) { + t.Fatal("first cue was rejected") + } + survivor := test_client(t, namespace) + wait_history(t, survivor, 1) + // Synchronize the cue timestamp before terminating the coordinator. + if _, err := survivor.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + owner.Close() + if survivor.Claim_sound(context.Background()) { + t.Fatal("coordinator exit reset the cooldown") + } + survivor.Publish("After takeover", time.Now()) + if _, err := survivor.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + joining := test_client(t, namespace) + wait_history(t, joining, 2) + survivor.Close() + joining.Close() + fresh := test_client(t, namespace) + if _, err := fresh.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + if len(fresh.Snapshot().Entries) != 0 || !fresh.Claim_sound(context.Background()) { + t.Fatal("history or cooldown survived the last reader") + } +} + +func TestCoordinatorExitFlushesNotificationToSurvivingReader(t *testing.T) { + namespace := fmt.Sprintf(".%x", random_id()) + owner := test_client(t, namespace) + if _, err := owner.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + survivor := test_client(t, namespace) + if _, err := survivor.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + owner.Publish("Immediately before exit", time.Now()) + owner.Close() + wait_history(t, survivor, 1) + if _, err := survivor.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + if got := survivor.Snapshot().Entries[0].Text; got != "Immediately before exit" { + t.Fatal(got) + } +} + +func TestMutedSoundRequestsUseSharedCooldown(t *testing.T) { + namespace := fmt.Sprintf(".%x", random_id()) + first, second := test_client(t, namespace), test_client(t, namespace) + played := make(chan struct{}, 4) + play := func(context.Context) { played <- struct{}{} } + first.Sound(play) + second.Sound(play) + select { + case <-played: + case <-time.After(3 * time.Second): + t.Fatal("muted notification cue was not delivered") + } + first.Close() + second.Close() + if len(played) != 0 { + t.Fatal("muted readers played multiple cues") + } +} + +func TestCanceledClaimDoesNotConsumeCooldownAndNamespacesStaySeparate(t *testing.T) { + namespace := fmt.Sprintf(".%x", random_id()) + first := test_client(t, namespace) + second := test_client(t, namespace+".other") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if first.Claim_sound(ctx) { + t.Fatal("canceled cue was accepted") + } + first.Publish("Private history", time.Now()) + if !first.Claim_sound(context.Background()) || !second.Claim_sound(context.Background()) { + t.Fatal("unrelated histories shared a cooldown") + } + if len(second.Snapshot().Entries) != 0 { + t.Fatal("history leaked to an unrelated group") + } +} + +func TestBrokerRejectsInvalidFramesAndAdmitsCueAfterCooldown(t *testing.T) { + namespace := fmt.Sprintf(".%x", random_id()) + c := test_client(t, namespace) + if _, err := c.exchange(context.Background(), false); err != nil { + t.Fatal(err) + } + for _, frame := range []string{ + `{"Version":2}`, + `{"Version":1,"Seed":{"Entries":[{"ID":"bad","Text":"bad"}]},"Claim":true}`, + `{"Version":1,"Seed":{"Entries":[],"LastSound":"2999-01-01T00:00:00Z"},"Claim":true}`, + } { + connection, err := net.DialTimeout("unix", c.address, time.Second) + if err != nil { + t.Fatal(err) + } + connection.SetDeadline(time.Now().Add(time.Second)) + fmt.Fprintln(connection, frame) + var result response + err = json.NewDecoder(connection).Decode(&result) + connection.Close() + if err == nil { + t.Fatal("invalid frame was accepted", frame) + } + } + c.brokerMu.Lock() + b := c.broker + c.brokerMu.Unlock() + b.mu.Lock() + b.state.LastSound = time.Now().Add(-SoundInterval) + b.mu.Unlock() + if !c.Claim_sound(context.Background()) { + t.Fatal("expired cooldown did not admit a cue") + } +} + +type processRequest struct { + Publish string + Claim bool + Stop bool +} + +type processResult struct { + State Snapshot + Allowed bool +} + +func TestNotificationProcess(t *testing.T) { + namespace := os.Getenv("FENRIR_NOTIFICATION_PROCESS") + if namespace == "" { + return + } + c := New(context.Background(), os.Geteuid(), namespace) + encoder := json.NewEncoder(os.Stdout) + decoder := json.NewDecoder(os.Stdin) + _ = encoder.Encode(processResult{}) + for { + var message processRequest + if decoder.Decode(&message) != nil || message.Stop { + c.Close() + os.Exit(0) + } + if message.Publish != "" { + c.Publish(message.Publish, time.Now()) + } + result, err := c.exchange(context.Background(), message.Claim) + if err != nil { + t.Fatal(err) + } + _ = encoder.Encode(processResult{State: result.State, Allowed: result.Allowed}) + } +} + +type notificationProcess struct { + command *exec.Cmd + input *json.Encoder + output *json.Decoder + stopped bool +} + +func start_process(t *testing.T, namespace string) *notificationProcess { + t.Helper() + command := exec.Command(os.Args[0], "-test.run=^TestNotificationProcess$") + command.Env = append(os.Environ(), "FENRIR_NOTIFICATION_PROCESS="+namespace) + command.Stderr = os.Stderr + input, err := command.StdinPipe() + if err != nil { + t.Fatal(err) + } + output, err := command.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := command.Start(); err != nil { + t.Fatal(err) + } + p := ¬ificationProcess{command: command, input: json.NewEncoder(input), output: json.NewDecoder(output)} + t.Cleanup(func() { p.stop(t) }) + var ready processResult + if err := p.output.Decode(&ready); err != nil { + t.Fatal(err) + } + return p +} + +func (p *notificationProcess) send(t *testing.T, message processRequest) processResult { + t.Helper() + if err := p.input.Encode(message); err != nil { + t.Fatal(err) + } + var result processResult + if err := p.output.Decode(&result); err != nil { + t.Fatal(err) + } + return result +} + +func (p *notificationProcess) stop(t *testing.T) { + t.Helper() + if p.stopped { + return + } + p.stopped = true + _ = p.input.Encode(processRequest{Stop: true}) + done := make(chan error, 1) + go func() { done <- p.command.Wait() }() + select { + case err := <-done: + if err != nil { + t.Error(err) + } + case <-time.After(5 * time.Second): + _ = p.command.Process.Kill() + <-done + t.Error("notification fixture did not exit") + } +} + +func TestFourProcessesShareHistoryAndCooldownAcrossCoordinatorExit(t *testing.T) { + for _, crash := range []bool{false, true} { + t.Run(fmt.Sprintf("crash=%v", crash), func(t *testing.T) { + check_four_processes(t, crash) + }) + } +} + +func check_four_processes(t *testing.T, crash bool) { + t.Helper() + namespace := fmt.Sprintf(".%x", random_id()) + processes := []*notificationProcess{start_process(t, namespace)} + want := processes[0].send(t, processRequest{Publish: "One targeted notification", Claim: true}) + if len(want.State.Entries) != 1 || !want.Allowed { + t.Fatal(want) + } + for range 3 { + processes = append(processes, start_process(t, namespace)) + } + for _, process := range processes[1:] { + got := process.send(t, processRequest{Claim: true}) + if !reflect.DeepEqual(got.State.Entries, want.State.Entries) || got.Allowed { + t.Fatal("a replica lost history or claimed another cue", got) + } + } + if crash { + processes[0].stopped = true + if err := processes[0].command.Process.Kill(); err != nil { + t.Fatal(err) + } + _ = processes[0].command.Wait() + } else { + processes[0].stop(t) + } + for _, process := range processes[1:] { + got := process.send(t, processRequest{Claim: true}) + if !reflect.DeepEqual(got.State.Entries, want.State.Entries) || got.Allowed { + t.Fatal("coordinator exit lost history or cooldown", got) + } + } + processes[1].send(t, processRequest{Publish: "Another targeted notification"}) + for _, process := range processes[1:] { + got := process.send(t, processRequest{}) + if len(got.State.Entries) != 2 || !strings.Contains(got.State.Entries[1].Text, "Another") { + t.Fatal(got) + } + } +}