Better notification organization.

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
+}

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 := &notificationProcess{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)
+		}
+	}
+}