feat(daemon): carry unsaved-state notices across restarts
When the daemon is shutting down and a session's terminal state can't be saved because the snapshot worker is saturated, there is no attached client left to toast. Persist the failure to a JSON-lines file (internal/adapters/noticefile) via the new ports.NoticeStore, and drain it at the next startup so the notice surfaces at the first client attach instead of being silently lost.
- commit
- db7ba0d99a5a0e22a6bf218c1a1d9ac62e4cbe83
- author
- brice · · 2026-07-21 04:47 UTC
- parent
- 2175cce
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +558 -1
A internal/adapters/noticefile/noticefile.go +94 -0
| @@ -0,0 +1,94 @@ | |||
| 1 | + | // Package noticefile persists undeliverable daemon notices as newline- | |
| 2 | + | // delimited JSON so they survive a daemon restart and can be shown at the | |
| 3 | + | // next client attach. | |
| 4 | + | package noticefile | |
| 5 | + | ||
| 6 | + | import ( | |
| 7 | + | "bufio" | |
| 8 | + | "encoding/json" | |
| 9 | + | "errors" | |
| 10 | + | "fmt" | |
| 11 | + | "os" | |
| 12 | + | "path/filepath" | |
| 13 | + | ||
| 14 | + | "github.com/bnema/vev/internal/domain" | |
| 15 | + | "github.com/bnema/vev/internal/ports" | |
| 16 | + | "github.com/bnema/vev/pkg/safedir" | |
| 17 | + | ) | |
| 18 | + | ||
| 19 | + | const fileName = "pending-notices.jsonl" | |
| 20 | + | ||
| 21 | + | // Store persists notices as JSON-lines under dir. | |
| 22 | + | type Store struct { | |
| 23 | + | dir string | |
| 24 | + | } | |
| 25 | + | ||
| 26 | + | var _ ports.NoticeStore = (*Store)(nil) | |
| 27 | + | ||
| 28 | + | // New creates a file-backed notice store rooted at dir. | |
| 29 | + | func New(dir string) *Store { | |
| 30 | + | return &Store{dir: dir} | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | func (s *Store) path() string { | |
| 34 | + | return filepath.Join(s.dir, fileName) | |
| 35 | + | } | |
| 36 | + | ||
| 37 | + | // Append writes n as one JSON line, creating the private state dir if needed. | |
| 38 | + | func (s *Store) Append(n domain.Notification) error { | |
| 39 | + | if err := safedir.EnsurePrivate(s.dir); err != nil { | |
| 40 | + | return fmt.Errorf("create notice dir: %w", err) | |
| 41 | + | } | |
| 42 | + | f, err := os.OpenFile(s.path(), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) | |
| 43 | + | if err != nil { | |
| 44 | + | return fmt.Errorf("open pending notices: %w", err) | |
| 45 | + | } | |
| 46 | + | defer func() { _ = f.Close() }() | |
| 47 | + | ||
| 48 | + | data, err := json.Marshal(n) | |
| 49 | + | if err != nil { | |
| 50 | + | return fmt.Errorf("marshal notice: %w", err) | |
| 51 | + | } | |
| 52 | + | data = append(data, '\n') | |
| 53 | + | if _, err := f.Write(data); err != nil { | |
| 54 | + | return fmt.Errorf("write notice: %w", err) | |
| 55 | + | } | |
| 56 | + | return nil | |
| 57 | + | } | |
| 58 | + | ||
| 59 | + | // Drain returns all stored notices and truncates the store. A missing file | |
| 60 | + | // means a fresh daemon has nothing pending, not an error. | |
| 61 | + | func (s *Store) Drain() ([]domain.Notification, error) { | |
| 62 | + | f, err := os.Open(s.path()) | |
| 63 | + | if errors.Is(err, os.ErrNotExist) { | |
| 64 | + | return nil, nil | |
| 65 | + | } | |
| 66 | + | if err != nil { | |
| 67 | + | return nil, fmt.Errorf("open pending notices: %w", err) | |
| 68 | + | } | |
| 69 | + | ||
| 70 | + | var notices []domain.Notification | |
| 71 | + | scanner := bufio.NewScanner(f) | |
| 72 | + | scanner.Buffer(make([]byte, 0, 64*1024), 4<<20) | |
| 73 | + | for scanner.Scan() { | |
| 74 | + | line := scanner.Bytes() | |
| 75 | + | if len(line) == 0 { | |
| 76 | + | continue | |
| 77 | + | } | |
| 78 | + | var n domain.Notification | |
| 79 | + | if err := json.Unmarshal(line, &n); err != nil { | |
| 80 | + | continue // corrupt or partial line must not abort the drain | |
| 81 | + | } | |
| 82 | + | notices = append(notices, n) | |
| 83 | + | } | |
| 84 | + | scanErr := scanner.Err() | |
| 85 | + | _ = f.Close() | |
| 86 | + | if scanErr != nil { | |
| 87 | + | return nil, fmt.Errorf("read pending notices: %w", scanErr) | |
| 88 | + | } | |
| 89 | + | ||
| 90 | + | if err := os.Truncate(s.path(), 0); err != nil { | |
| 91 | + | return nil, fmt.Errorf("truncate pending notices: %w", err) | |
| 92 | + | } | |
| 93 | + | return notices, nil | |
| 94 | + | } | |
| @@ -0,0 +1,94 @@ | |||||
| 1 | + | // Package noticefile persists undeliverable daemon notices as newline- | |||
| 2 | + | // delimited JSON so they survive a daemon restart and can be shown at the | |||
| 3 | + | // next client attach. | |||
| 4 | + | package noticefile | |||
| 5 | + | ||||
| 6 | + | import ( | |||
| 7 | + | "bufio" | |||
| 8 | + | "encoding/json" | |||
| 9 | + | "errors" | |||
| 10 | + | "fmt" | |||
| 11 | + | "os" | |||
| 12 | + | "path/filepath" | |||
| 13 | + | ||||
| 14 | + | "github.com/bnema/vev/internal/domain" | |||
| 15 | + | "github.com/bnema/vev/internal/ports" | |||
| 16 | + | "github.com/bnema/vev/pkg/safedir" | |||
| 17 | + | ) | |||
| 18 | + | ||||
| 19 | + | const fileName = "pending-notices.jsonl" | |||
| 20 | + | ||||
| 21 | + | // Store persists notices as JSON-lines under dir. | |||
| 22 | + | type Store struct { | |||
| 23 | + | dir string | |||
| 24 | + | } | |||
| 25 | + | ||||
| 26 | + | var _ ports.NoticeStore = (*Store)(nil) | |||
| 27 | + | ||||
| 28 | + | // New creates a file-backed notice store rooted at dir. | |||
| 29 | + | func New(dir string) *Store { | |||
| 30 | + | return &Store{dir: dir} | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | func (s *Store) path() string { | |||
| 34 | + | return filepath.Join(s.dir, fileName) | |||
| 35 | + | } | |||
| 36 | + | ||||
| 37 | + | // Append writes n as one JSON line, creating the private state dir if needed. | |||
| 38 | + | func (s *Store) Append(n domain.Notification) error { | |||
| 39 | + | if err := safedir.EnsurePrivate(s.dir); err != nil { | |||
| 40 | + | return fmt.Errorf("create notice dir: %w", err) | |||
| 41 | + | } | |||
| 42 | + | f, err := os.OpenFile(s.path(), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) | |||
| 43 | + | if err != nil { | |||
| 44 | + | return fmt.Errorf("open pending notices: %w", err) | |||
| 45 | + | } | |||
| 46 | + | defer func() { _ = f.Close() }() | |||
| 47 | + | ||||
| 48 | + | data, err := json.Marshal(n) | |||
| 49 | + | if err != nil { | |||
| 50 | + | return fmt.Errorf("marshal notice: %w", err) | |||
| 51 | + | } | |||
| 52 | + | data = append(data, '\n') | |||
| 53 | + | if _, err := f.Write(data); err != nil { | |||
| 54 | + | return fmt.Errorf("write notice: %w", err) | |||
| 55 | + | } | |||
| 56 | + | return nil | |||
| 57 | + | } | |||
| 58 | + | ||||
| 59 | + | // Drain returns all stored notices and truncates the store. A missing file | |||
| 60 | + | // means a fresh daemon has nothing pending, not an error. | |||
| 61 | + | func (s *Store) Drain() ([]domain.Notification, error) { | |||
| 62 | + | f, err := os.Open(s.path()) | |||
| 63 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 64 | + | return nil, nil | |||
| 65 | + | } | |||
| 66 | + | if err != nil { | |||
| 67 | + | return nil, fmt.Errorf("open pending notices: %w", err) | |||
| 68 | + | } | |||
| 69 | + | ||||
| 70 | + | var notices []domain.Notification | |||
| 71 | + | scanner := bufio.NewScanner(f) | |||
| 72 | + | scanner.Buffer(make([]byte, 0, 64*1024), 4<<20) | |||
| 73 | + | for scanner.Scan() { | |||
| 74 | + | line := scanner.Bytes() | |||
| 75 | + | if len(line) == 0 { | |||
| 76 | + | continue | |||
| 77 | + | } | |||
| 78 | + | var n domain.Notification | |||
| 79 | + | if err := json.Unmarshal(line, &n); err != nil { | |||
| 80 | + | continue // corrupt or partial line must not abort the drain | |||
| 81 | + | } | |||
| 82 | + | notices = append(notices, n) | |||
| 83 | + | } | |||
| 84 | + | scanErr := scanner.Err() | |||
| 85 | + | _ = f.Close() | |||
| 86 | + | if scanErr != nil { | |||
| 87 | + | return nil, fmt.Errorf("read pending notices: %w", scanErr) | |||
| 88 | + | } | |||
| 89 | + | ||||
| 90 | + | if err := os.Truncate(s.path(), 0); err != nil { | |||
| 91 | + | return nil, fmt.Errorf("truncate pending notices: %w", err) | |||
| 92 | + | } | |||
| 93 | + | return notices, nil | |||
| 94 | + | } | |||
A internal/adapters/noticefile/noticefile_test.go +139 -0
| @@ -0,0 +1,139 @@ | |||
| 1 | + | package noticefile | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "os" | |
| 5 | + | "path/filepath" | |
| 6 | + | "testing" | |
| 7 | + | "time" | |
| 8 | + | ||
| 9 | + | "github.com/bnema/vev/internal/domain" | |
| 10 | + | ) | |
| 11 | + | ||
| 12 | + | func TestStoreAppendDrainRoundTrip(t *testing.T) { | |
| 13 | + | t.Parallel() | |
| 14 | + | dir := filepath.Join(t.TempDir(), "vev") | |
| 15 | + | store := New(dir) | |
| 16 | + | ||
| 17 | + | first := domain.Notification{ | |
| 18 | + | Code: domain.NoticeSnapshotWrite, | |
| 19 | + | Severity: domain.NoticeError, | |
| 20 | + | Message: "session foo shut down without saving terminal state", | |
| 21 | + | Details: "boom: disk full", | |
| 22 | + | Time: time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC), | |
| 23 | + | Count: 1, | |
| 24 | + | SessionID: "sess-1", | |
| 25 | + | } | |
| 26 | + | second := domain.Notification{ | |
| 27 | + | Code: domain.NoticeSnapshotRestore, | |
| 28 | + | Severity: domain.NoticeWarn, | |
| 29 | + | Message: "session bar could not be restored", | |
| 30 | + | Time: time.Date(2026, 7, 20, 10, 5, 0, 0, time.UTC), | |
| 31 | + | Count: 2, | |
| 32 | + | } | |
| 33 | + | ||
| 34 | + | if err := store.Append(first); err != nil { | |
| 35 | + | t.Fatalf("Append first: %v", err) | |
| 36 | + | } | |
| 37 | + | if err := store.Append(second); err != nil { | |
| 38 | + | t.Fatalf("Append second: %v", err) | |
| 39 | + | } | |
| 40 | + | ||
| 41 | + | got, err := store.Drain() | |
| 42 | + | if err != nil { | |
| 43 | + | t.Fatalf("Drain: %v", err) | |
| 44 | + | } | |
| 45 | + | if len(got) != 2 { | |
| 46 | + | t.Fatalf("Drain len = %d, want 2", len(got)) | |
| 47 | + | } | |
| 48 | + | if !got[0].Time.Equal(first.Time) || got[0].Message != first.Message || got[0].Code != first.Code || | |
| 49 | + | got[0].Severity != first.Severity || got[0].Details != first.Details || got[0].Count != first.Count || | |
| 50 | + | got[0].SessionID != first.SessionID { | |
| 51 | + | t.Fatalf("first notice round-trip mismatch: got %#v, want %#v", got[0], first) | |
| 52 | + | } | |
| 53 | + | if !got[1].Time.Equal(second.Time) || got[1].Message != second.Message || got[1].Code != second.Code || | |
| 54 | + | got[1].Severity != second.Severity || got[1].Count != second.Count { | |
| 55 | + | t.Fatalf("second notice round-trip mismatch: got %#v, want %#v", got[1], second) | |
| 56 | + | } | |
| 57 | + | ||
| 58 | + | again, err := store.Drain() | |
| 59 | + | if err != nil { | |
| 60 | + | t.Fatalf("Drain again: %v", err) | |
| 61 | + | } | |
| 62 | + | if len(again) != 0 { | |
| 63 | + | t.Fatalf("Drain again len = %d, want 0", len(again)) | |
| 64 | + | } | |
| 65 | + | } | |
| 66 | + | ||
| 67 | + | func TestStoreDrainMissingFileReturnsNil(t *testing.T) { | |
| 68 | + | t.Parallel() | |
| 69 | + | dir := t.TempDir() | |
| 70 | + | store := New(dir) | |
| 71 | + | ||
| 72 | + | got, err := store.Drain() | |
| 73 | + | if err != nil { | |
| 74 | + | t.Fatalf("Drain: %v", err) | |
| 75 | + | } | |
| 76 | + | if got != nil { | |
| 77 | + | t.Fatalf("Drain on missing file = %#v, want nil", got) | |
| 78 | + | } | |
| 79 | + | } | |
| 80 | + | ||
| 81 | + | func TestStoreDrainSkipsGarbageLines(t *testing.T) { | |
| 82 | + | t.Parallel() | |
| 83 | + | dir := t.TempDir() | |
| 84 | + | store := New(dir) | |
| 85 | + | path := filepath.Join(dir, "pending-notices.jsonl") | |
| 86 | + | if err := os.MkdirAll(dir, 0o700); err != nil { | |
| 87 | + | t.Fatalf("MkdirAll: %v", err) | |
| 88 | + | } | |
| 89 | + | ||
| 90 | + | good := `{"Code":9,"Severity":2,"Message":"first"}` + "\n" | |
| 91 | + | garbage := "not json at all\n" | |
| 92 | + | good2 := `{"Code":10,"Severity":1,"Message":"second"}` + "\n" | |
| 93 | + | if err := os.WriteFile(path, []byte(good+garbage+good2), 0o600); err != nil { | |
| 94 | + | t.Fatalf("WriteFile: %v", err) | |
| 95 | + | } | |
| 96 | + | ||
| 97 | + | got, err := store.Drain() | |
| 98 | + | if err != nil { | |
| 99 | + | t.Fatalf("Drain: %v", err) | |
| 100 | + | } | |
| 101 | + | if len(got) != 2 { | |
| 102 | + | t.Fatalf("Drain len = %d, want 2 (garbage line skipped)", len(got)) | |
| 103 | + | } | |
| 104 | + | if got[0].Message != "first" || got[1].Message != "second" { | |
| 105 | + | t.Fatalf("Drain messages = %q, %q; want first, second", got[0].Message, got[1].Message) | |
| 106 | + | } | |
| 107 | + | ||
| 108 | + | data, err := os.ReadFile(path) | |
| 109 | + | if err != nil { | |
| 110 | + | t.Fatalf("ReadFile after drain: %v", err) | |
| 111 | + | } | |
| 112 | + | if len(data) != 0 { | |
| 113 | + | t.Fatalf("file not truncated after Drain, got %d bytes", len(data)) | |
| 114 | + | } | |
| 115 | + | } | |
| 116 | + | ||
| 117 | + | func TestStoreAppendPrivatizesDir(t *testing.T) { | |
| 118 | + | t.Parallel() | |
| 119 | + | dir := filepath.Join(t.TempDir(), "vev") | |
| 120 | + | store := New(dir) | |
| 121 | + | ||
| 122 | + | if err := store.Append(domain.Notification{Message: "x"}); err != nil { | |
| 123 | + | t.Fatalf("Append: %v", err) | |
| 124 | + | } | |
| 125 | + | info, err := os.Stat(dir) | |
| 126 | + | if err != nil { | |
| 127 | + | t.Fatalf("Stat dir: %v", err) | |
| 128 | + | } | |
| 129 | + | if got := info.Mode().Perm(); got != 0o700 { | |
| 130 | + | t.Fatalf("dir mode = %v, want 0700", got) | |
| 131 | + | } | |
| 132 | + | fi, err := os.Stat(filepath.Join(dir, "pending-notices.jsonl")) | |
| 133 | + | if err != nil { | |
| 134 | + | t.Fatalf("Stat file: %v", err) | |
| 135 | + | } | |
| 136 | + | if got := fi.Mode().Perm(); got != 0o600 { | |
| 137 | + | t.Fatalf("file mode = %v, want 0600", got) | |
| 138 | + | } | |
| 139 | + | } | |
| @@ -0,0 +1,139 @@ | |||||
| 1 | + | package noticefile | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "os" | |||
| 5 | + | "path/filepath" | |||
| 6 | + | "testing" | |||
| 7 | + | "time" | |||
| 8 | + | ||||
| 9 | + | "github.com/bnema/vev/internal/domain" | |||
| 10 | + | ) | |||
| 11 | + | ||||
| 12 | + | func TestStoreAppendDrainRoundTrip(t *testing.T) { | |||
| 13 | + | t.Parallel() | |||
| 14 | + | dir := filepath.Join(t.TempDir(), "vev") | |||
| 15 | + | store := New(dir) | |||
| 16 | + | ||||
| 17 | + | first := domain.Notification{ | |||
| 18 | + | Code: domain.NoticeSnapshotWrite, | |||
| 19 | + | Severity: domain.NoticeError, | |||
| 20 | + | Message: "session foo shut down without saving terminal state", | |||
| 21 | + | Details: "boom: disk full", | |||
| 22 | + | Time: time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC), | |||
| 23 | + | Count: 1, | |||
| 24 | + | SessionID: "sess-1", | |||
| 25 | + | } | |||
| 26 | + | second := domain.Notification{ | |||
| 27 | + | Code: domain.NoticeSnapshotRestore, | |||
| 28 | + | Severity: domain.NoticeWarn, | |||
| 29 | + | Message: "session bar could not be restored", | |||
| 30 | + | Time: time.Date(2026, 7, 20, 10, 5, 0, 0, time.UTC), | |||
| 31 | + | Count: 2, | |||
| 32 | + | } | |||
| 33 | + | ||||
| 34 | + | if err := store.Append(first); err != nil { | |||
| 35 | + | t.Fatalf("Append first: %v", err) | |||
| 36 | + | } | |||
| 37 | + | if err := store.Append(second); err != nil { | |||
| 38 | + | t.Fatalf("Append second: %v", err) | |||
| 39 | + | } | |||
| 40 | + | ||||
| 41 | + | got, err := store.Drain() | |||
| 42 | + | if err != nil { | |||
| 43 | + | t.Fatalf("Drain: %v", err) | |||
| 44 | + | } | |||
| 45 | + | if len(got) != 2 { | |||
| 46 | + | t.Fatalf("Drain len = %d, want 2", len(got)) | |||
| 47 | + | } | |||
| 48 | + | if !got[0].Time.Equal(first.Time) || got[0].Message != first.Message || got[0].Code != first.Code || | |||
| 49 | + | got[0].Severity != first.Severity || got[0].Details != first.Details || got[0].Count != first.Count || | |||
| 50 | + | got[0].SessionID != first.SessionID { | |||
| 51 | + | t.Fatalf("first notice round-trip mismatch: got %#v, want %#v", got[0], first) | |||
| 52 | + | } | |||
| 53 | + | if !got[1].Time.Equal(second.Time) || got[1].Message != second.Message || got[1].Code != second.Code || | |||
| 54 | + | got[1].Severity != second.Severity || got[1].Count != second.Count { | |||
| 55 | + | t.Fatalf("second notice round-trip mismatch: got %#v, want %#v", got[1], second) | |||
| 56 | + | } | |||
| 57 | + | ||||
| 58 | + | again, err := store.Drain() | |||
| 59 | + | if err != nil { | |||
| 60 | + | t.Fatalf("Drain again: %v", err) | |||
| 61 | + | } | |||
| 62 | + | if len(again) != 0 { | |||
| 63 | + | t.Fatalf("Drain again len = %d, want 0", len(again)) | |||
| 64 | + | } | |||
| 65 | + | } | |||
| 66 | + | ||||
| 67 | + | func TestStoreDrainMissingFileReturnsNil(t *testing.T) { | |||
| 68 | + | t.Parallel() | |||
| 69 | + | dir := t.TempDir() | |||
| 70 | + | store := New(dir) | |||
| 71 | + | ||||
| 72 | + | got, err := store.Drain() | |||
| 73 | + | if err != nil { | |||
| 74 | + | t.Fatalf("Drain: %v", err) | |||
| 75 | + | } | |||
| 76 | + | if got != nil { | |||
| 77 | + | t.Fatalf("Drain on missing file = %#v, want nil", got) | |||
| 78 | + | } | |||
| 79 | + | } | |||
| 80 | + | ||||
| 81 | + | func TestStoreDrainSkipsGarbageLines(t *testing.T) { | |||
| 82 | + | t.Parallel() | |||
| 83 | + | dir := t.TempDir() | |||
| 84 | + | store := New(dir) | |||
| 85 | + | path := filepath.Join(dir, "pending-notices.jsonl") | |||
| 86 | + | if err := os.MkdirAll(dir, 0o700); err != nil { | |||
| 87 | + | t.Fatalf("MkdirAll: %v", err) | |||
| 88 | + | } | |||
| 89 | + | ||||
| 90 | + | good := `{"Code":9,"Severity":2,"Message":"first"}` + "\n" | |||
| 91 | + | garbage := "not json at all\n" | |||
| 92 | + | good2 := `{"Code":10,"Severity":1,"Message":"second"}` + "\n" | |||
| 93 | + | if err := os.WriteFile(path, []byte(good+garbage+good2), 0o600); err != nil { | |||
| 94 | + | t.Fatalf("WriteFile: %v", err) | |||
| 95 | + | } | |||
| 96 | + | ||||
| 97 | + | got, err := store.Drain() | |||
| 98 | + | if err != nil { | |||
| 99 | + | t.Fatalf("Drain: %v", err) | |||
| 100 | + | } | |||
| 101 | + | if len(got) != 2 { | |||
| 102 | + | t.Fatalf("Drain len = %d, want 2 (garbage line skipped)", len(got)) | |||
| 103 | + | } | |||
| 104 | + | if got[0].Message != "first" || got[1].Message != "second" { | |||
| 105 | + | t.Fatalf("Drain messages = %q, %q; want first, second", got[0].Message, got[1].Message) | |||
| 106 | + | } | |||
| 107 | + | ||||
| 108 | + | data, err := os.ReadFile(path) | |||
| 109 | + | if err != nil { | |||
| 110 | + | t.Fatalf("ReadFile after drain: %v", err) | |||
| 111 | + | } | |||
| 112 | + | if len(data) != 0 { | |||
| 113 | + | t.Fatalf("file not truncated after Drain, got %d bytes", len(data)) | |||
| 114 | + | } | |||
| 115 | + | } | |||
| 116 | + | ||||
| 117 | + | func TestStoreAppendPrivatizesDir(t *testing.T) { | |||
| 118 | + | t.Parallel() | |||
| 119 | + | dir := filepath.Join(t.TempDir(), "vev") | |||
| 120 | + | store := New(dir) | |||
| 121 | + | ||||
| 122 | + | if err := store.Append(domain.Notification{Message: "x"}); err != nil { | |||
| 123 | + | t.Fatalf("Append: %v", err) | |||
| 124 | + | } | |||
| 125 | + | info, err := os.Stat(dir) | |||
| 126 | + | if err != nil { | |||
| 127 | + | t.Fatalf("Stat dir: %v", err) | |||
| 128 | + | } | |||
| 129 | + | if got := info.Mode().Perm(); got != 0o700 { | |||
| 130 | + | t.Fatalf("dir mode = %v, want 0700", got) | |||
| 131 | + | } | |||
| 132 | + | fi, err := os.Stat(filepath.Join(dir, "pending-notices.jsonl")) | |||
| 133 | + | if err != nil { | |||
| 134 | + | t.Fatalf("Stat file: %v", err) | |||
| 135 | + | } | |||
| 136 | + | if got := fi.Mode().Perm(); got != 0o600 { | |||
| 137 | + | t.Fatalf("file mode = %v, want 0600", got) | |||
| 138 | + | } | |||
| 139 | + | } | |||
M internal/app/run.go +2 -0
| @@ -32,6 +32,7 @@ import ( | |||
| 32 | 32 | "github.com/bnema/vev/internal/adapters/config" | |
| 33 | 33 | "github.com/bnema/vev/internal/adapters/dgram" | |
| 34 | 34 | "github.com/bnema/vev/internal/adapters/ipc" | |
| 35 | + | "github.com/bnema/vev/internal/adapters/noticefile" | |
| 35 | 36 | "github.com/bnema/vev/internal/adapters/observability" | |
| 36 | 37 | "github.com/bnema/vev/internal/adapters/pty" | |
| 37 | 38 | remoteadapter "github.com/bnema/vev/internal/adapters/remote" | |
| @@ -450,6 +451,7 @@ func runDaemon() (retErr error) { | |||
| 450 | 451 | daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New())) | |
| 451 | 452 | daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome)) | |
| 452 | 453 | daemonOpts = append(daemonOpts, daemon.WithSnapshotStore(snapshotadapter.NewStore(snapshotDir()))) | |
| 454 | + | daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir()))) | |
| 453 | 455 | storePath := persist.StorePath(platform.StateDir()) | |
| 454 | 456 | var storeErr error | |
| 455 | 457 | if store, err := kv.Open(storePath); err != nil { | |
| @@ -32,6 +32,7 @@ import ( | |||||
| 32 | "github.com/bnema/vev/internal/adapters/config" | 32 | "github.com/bnema/vev/internal/adapters/config" | ||
| 33 | "github.com/bnema/vev/internal/adapters/dgram" | 33 | "github.com/bnema/vev/internal/adapters/dgram" | ||
| 34 | "github.com/bnema/vev/internal/adapters/ipc" | 34 | "github.com/bnema/vev/internal/adapters/ipc" | ||
| 35 | + | "github.com/bnema/vev/internal/adapters/noticefile" | |||
| 35 | "github.com/bnema/vev/internal/adapters/observability" | 36 | "github.com/bnema/vev/internal/adapters/observability" | ||
| 36 | "github.com/bnema/vev/internal/adapters/pty" | 37 | "github.com/bnema/vev/internal/adapters/pty" | ||
| 37 | remoteadapter "github.com/bnema/vev/internal/adapters/remote" | 38 | remoteadapter "github.com/bnema/vev/internal/adapters/remote" | ||
| @@ -450,6 +451,7 @@ func runDaemon() (retErr error) { | |||||
| 450 | daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New())) | 451 | daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New())) | ||
| 451 | daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome)) | 452 | daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome)) | ||
| 452 | daemonOpts = append(daemonOpts, daemon.WithSnapshotStore(snapshotadapter.NewStore(snapshotDir()))) | 453 | daemonOpts = append(daemonOpts, daemon.WithSnapshotStore(snapshotadapter.NewStore(snapshotDir()))) | ||
| 454 | + | daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir()))) | |||
| 453 | storePath := persist.StorePath(platform.StateDir()) | 455 | storePath := persist.StorePath(platform.StateDir()) | ||
| 454 | var storeErr error | 456 | var storeErr error | ||
| 455 | if store, err := kv.Open(storePath); err != nil { | 457 | if store, err := kv.Open(storePath); err != nil { | ||
A internal/ports/mocks/notice_mock.go +143 -0
| @@ -0,0 +1,143 @@ | |||
| 1 | + | // Code generated by mockery; DO NOT EDIT. | |
| 2 | + | // github.com/vektra/mockery | |
| 3 | + | // template: testify | |
| 4 | + | ||
| 5 | + | package portsmocks | |
| 6 | + | ||
| 7 | + | import ( | |
| 8 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | + | mock "github.com/stretchr/testify/mock" | |
| 10 | + | ) | |
| 11 | + | ||
| 12 | + | // NewMockNoticeStore creates a new instance of MockNoticeStore. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. | |
| 13 | + | // The first argument is typically a *testing.T value. | |
| 14 | + | func NewMockNoticeStore(t interface { | |
| 15 | + | mock.TestingT | |
| 16 | + | Cleanup(func()) | |
| 17 | + | }) *MockNoticeStore { | |
| 18 | + | mock := &MockNoticeStore{} | |
| 19 | + | mock.Mock.Test(t) | |
| 20 | + | ||
| 21 | + | t.Cleanup(func() { mock.AssertExpectations(t) }) | |
| 22 | + | ||
| 23 | + | return mock | |
| 24 | + | } | |
| 25 | + | ||
| 26 | + | // MockNoticeStore is an autogenerated mock type for the NoticeStore type | |
| 27 | + | type MockNoticeStore struct { | |
| 28 | + | mock.Mock | |
| 29 | + | } | |
| 30 | + | ||
| 31 | + | type MockNoticeStore_Expecter struct { | |
| 32 | + | mock *mock.Mock | |
| 33 | + | } | |
| 34 | + | ||
| 35 | + | func (_m *MockNoticeStore) EXPECT() *MockNoticeStore_Expecter { | |
| 36 | + | return &MockNoticeStore_Expecter{mock: &_m.Mock} | |
| 37 | + | } | |
| 38 | + | ||
| 39 | + | // Append provides a mock function for the type MockNoticeStore | |
| 40 | + | func (_mock *MockNoticeStore) Append(n domain.Notification) error { | |
| 41 | + | ret := _mock.Called(n) | |
| 42 | + | ||
| 43 | + | if len(ret) == 0 { | |
| 44 | + | panic("no return value specified for Append") | |
| 45 | + | } | |
| 46 | + | ||
| 47 | + | var r0 error | |
| 48 | + | if returnFunc, ok := ret.Get(0).(func(domain.Notification) error); ok { | |
| 49 | + | r0 = returnFunc(n) | |
| 50 | + | } else { | |
| 51 | + | r0 = ret.Error(0) | |
| 52 | + | } | |
| 53 | + | return r0 | |
| 54 | + | } | |
| 55 | + | ||
| 56 | + | // MockNoticeStore_Append_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Append' | |
| 57 | + | type MockNoticeStore_Append_Call struct { | |
| 58 | + | *mock.Call | |
| 59 | + | } | |
| 60 | + | ||
| 61 | + | // Append is a helper method to define mock.On call | |
| 62 | + | // - n domain.Notification | |
| 63 | + | func (_e *MockNoticeStore_Expecter) Append(n any) *MockNoticeStore_Append_Call { | |
| 64 | + | return &MockNoticeStore_Append_Call{Call: _e.mock.On("Append", n)} | |
| 65 | + | } | |
| 66 | + | ||
| 67 | + | func (_c *MockNoticeStore_Append_Call) Run(run func(n domain.Notification)) *MockNoticeStore_Append_Call { | |
| 68 | + | _c.Call.Run(func(args mock.Arguments) { | |
| 69 | + | var arg0 domain.Notification | |
| 70 | + | if args[0] != nil { | |
| 71 | + | arg0 = args[0].(domain.Notification) | |
| 72 | + | } | |
| 73 | + | run( | |
| 74 | + | arg0, | |
| 75 | + | ) | |
| 76 | + | }) | |
| 77 | + | return _c | |
| 78 | + | } | |
| 79 | + | ||
| 80 | + | func (_c *MockNoticeStore_Append_Call) Return(err error) *MockNoticeStore_Append_Call { | |
| 81 | + | _c.Call.Return(err) | |
| 82 | + | return _c | |
| 83 | + | } | |
| 84 | + | ||
| 85 | + | func (_c *MockNoticeStore_Append_Call) RunAndReturn(run func(n domain.Notification) error) *MockNoticeStore_Append_Call { | |
| 86 | + | _c.Call.Return(run) | |
| 87 | + | return _c | |
| 88 | + | } | |
| 89 | + | ||
| 90 | + | // Drain provides a mock function for the type MockNoticeStore | |
| 91 | + | func (_mock *MockNoticeStore) Drain() ([]domain.Notification, error) { | |
| 92 | + | ret := _mock.Called() | |
| 93 | + | ||
| 94 | + | if len(ret) == 0 { | |
| 95 | + | panic("no return value specified for Drain") | |
| 96 | + | } | |
| 97 | + | ||
| 98 | + | var r0 []domain.Notification | |
| 99 | + | var r1 error | |
| 100 | + | if returnFunc, ok := ret.Get(0).(func() ([]domain.Notification, error)); ok { | |
| 101 | + | return returnFunc() | |
| 102 | + | } | |
| 103 | + | if returnFunc, ok := ret.Get(0).(func() []domain.Notification); ok { | |
| 104 | + | r0 = returnFunc() | |
| 105 | + | } else { | |
| 106 | + | if ret.Get(0) != nil { | |
| 107 | + | r0 = ret.Get(0).([]domain.Notification) | |
| 108 | + | } | |
| 109 | + | } | |
| 110 | + | if returnFunc, ok := ret.Get(1).(func() error); ok { | |
| 111 | + | r1 = returnFunc() | |
| 112 | + | } else { | |
| 113 | + | r1 = ret.Error(1) | |
| 114 | + | } | |
| 115 | + | return r0, r1 | |
| 116 | + | } | |
| 117 | + | ||
| 118 | + | // MockNoticeStore_Drain_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Drain' | |
| 119 | + | type MockNoticeStore_Drain_Call struct { | |
| 120 | + | *mock.Call | |
| 121 | + | } | |
| 122 | + | ||
| 123 | + | // Drain is a helper method to define mock.On call | |
| 124 | + | func (_e *MockNoticeStore_Expecter) Drain() *MockNoticeStore_Drain_Call { | |
| 125 | + | return &MockNoticeStore_Drain_Call{Call: _e.mock.On("Drain")} | |
| 126 | + | } | |
| 127 | + | ||
| 128 | + | func (_c *MockNoticeStore_Drain_Call) Run(run func()) *MockNoticeStore_Drain_Call { | |
| 129 | + | _c.Call.Run(func(args mock.Arguments) { | |
| 130 | + | run() | |
| 131 | + | }) | |
| 132 | + | return _c | |
| 133 | + | } | |
| 134 | + | ||
| 135 | + | func (_c *MockNoticeStore_Drain_Call) Return(notifications []domain.Notification, err error) *MockNoticeStore_Drain_Call { | |
| 136 | + | _c.Call.Return(notifications, err) | |
| 137 | + | return _c | |
| 138 | + | } | |
| 139 | + | ||
| 140 | + | func (_c *MockNoticeStore_Drain_Call) RunAndReturn(run func() ([]domain.Notification, error)) *MockNoticeStore_Drain_Call { | |
| 141 | + | _c.Call.Return(run) | |
| 142 | + | return _c | |
| 143 | + | } | |
| @@ -0,0 +1,143 @@ | |||||
| 1 | + | // Code generated by mockery; DO NOT EDIT. | |||
| 2 | + | // github.com/vektra/mockery | |||
| 3 | + | // template: testify | |||
| 4 | + | ||||
| 5 | + | package portsmocks | |||
| 6 | + | ||||
| 7 | + | import ( | |||
| 8 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | + | mock "github.com/stretchr/testify/mock" | |||
| 10 | + | ) | |||
| 11 | + | ||||
| 12 | + | // NewMockNoticeStore creates a new instance of MockNoticeStore. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. | |||
| 13 | + | // The first argument is typically a *testing.T value. | |||
| 14 | + | func NewMockNoticeStore(t interface { | |||
| 15 | + | mock.TestingT | |||
| 16 | + | Cleanup(func()) | |||
| 17 | + | }) *MockNoticeStore { | |||
| 18 | + | mock := &MockNoticeStore{} | |||
| 19 | + | mock.Mock.Test(t) | |||
| 20 | + | ||||
| 21 | + | t.Cleanup(func() { mock.AssertExpectations(t) }) | |||
| 22 | + | ||||
| 23 | + | return mock | |||
| 24 | + | } | |||
| 25 | + | ||||
| 26 | + | // MockNoticeStore is an autogenerated mock type for the NoticeStore type | |||
| 27 | + | type MockNoticeStore struct { | |||
| 28 | + | mock.Mock | |||
| 29 | + | } | |||
| 30 | + | ||||
| 31 | + | type MockNoticeStore_Expecter struct { | |||
| 32 | + | mock *mock.Mock | |||
| 33 | + | } | |||
| 34 | + | ||||
| 35 | + | func (_m *MockNoticeStore) EXPECT() *MockNoticeStore_Expecter { | |||
| 36 | + | return &MockNoticeStore_Expecter{mock: &_m.Mock} | |||
| 37 | + | } | |||
| 38 | + | ||||
| 39 | + | // Append provides a mock function for the type MockNoticeStore | |||
| 40 | + | func (_mock *MockNoticeStore) Append(n domain.Notification) error { | |||
| 41 | + | ret := _mock.Called(n) | |||
| 42 | + | ||||
| 43 | + | if len(ret) == 0 { | |||
| 44 | + | panic("no return value specified for Append") | |||
| 45 | + | } | |||
| 46 | + | ||||
| 47 | + | var r0 error | |||
| 48 | + | if returnFunc, ok := ret.Get(0).(func(domain.Notification) error); ok { | |||
| 49 | + | r0 = returnFunc(n) | |||
| 50 | + | } else { | |||
| 51 | + | r0 = ret.Error(0) | |||
| 52 | + | } | |||
| 53 | + | return r0 | |||
| 54 | + | } | |||
| 55 | + | ||||
| 56 | + | // MockNoticeStore_Append_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Append' | |||
| 57 | + | type MockNoticeStore_Append_Call struct { | |||
| 58 | + | *mock.Call | |||
| 59 | + | } | |||
| 60 | + | ||||
| 61 | + | // Append is a helper method to define mock.On call | |||
| 62 | + | // - n domain.Notification | |||
| 63 | + | func (_e *MockNoticeStore_Expecter) Append(n any) *MockNoticeStore_Append_Call { | |||
| 64 | + | return &MockNoticeStore_Append_Call{Call: _e.mock.On("Append", n)} | |||
| 65 | + | } | |||
| 66 | + | ||||
| 67 | + | func (_c *MockNoticeStore_Append_Call) Run(run func(n domain.Notification)) *MockNoticeStore_Append_Call { | |||
| 68 | + | _c.Call.Run(func(args mock.Arguments) { | |||
| 69 | + | var arg0 domain.Notification | |||
| 70 | + | if args[0] != nil { | |||
| 71 | + | arg0 = args[0].(domain.Notification) | |||
| 72 | + | } | |||
| 73 | + | run( | |||
| 74 | + | arg0, | |||
| 75 | + | ) | |||
| 76 | + | }) | |||
| 77 | + | return _c | |||
| 78 | + | } | |||
| 79 | + | ||||
| 80 | + | func (_c *MockNoticeStore_Append_Call) Return(err error) *MockNoticeStore_Append_Call { | |||
| 81 | + | _c.Call.Return(err) | |||
| 82 | + | return _c | |||
| 83 | + | } | |||
| 84 | + | ||||
| 85 | + | func (_c *MockNoticeStore_Append_Call) RunAndReturn(run func(n domain.Notification) error) *MockNoticeStore_Append_Call { | |||
| 86 | + | _c.Call.Return(run) | |||
| 87 | + | return _c | |||
| 88 | + | } | |||
| 89 | + | ||||
| 90 | + | // Drain provides a mock function for the type MockNoticeStore | |||
| 91 | + | func (_mock *MockNoticeStore) Drain() ([]domain.Notification, error) { | |||
| 92 | + | ret := _mock.Called() | |||
| 93 | + | ||||
| 94 | + | if len(ret) == 0 { | |||
| 95 | + | panic("no return value specified for Drain") | |||
| 96 | + | } | |||
| 97 | + | ||||
| 98 | + | var r0 []domain.Notification | |||
| 99 | + | var r1 error | |||
| 100 | + | if returnFunc, ok := ret.Get(0).(func() ([]domain.Notification, error)); ok { | |||
| 101 | + | return returnFunc() | |||
| 102 | + | } | |||
| 103 | + | if returnFunc, ok := ret.Get(0).(func() []domain.Notification); ok { | |||
| 104 | + | r0 = returnFunc() | |||
| 105 | + | } else { | |||
| 106 | + | if ret.Get(0) != nil { | |||
| 107 | + | r0 = ret.Get(0).([]domain.Notification) | |||
| 108 | + | } | |||
| 109 | + | } | |||
| 110 | + | if returnFunc, ok := ret.Get(1).(func() error); ok { | |||
| 111 | + | r1 = returnFunc() | |||
| 112 | + | } else { | |||
| 113 | + | r1 = ret.Error(1) | |||
| 114 | + | } | |||
| 115 | + | return r0, r1 | |||
| 116 | + | } | |||
| 117 | + | ||||
| 118 | + | // MockNoticeStore_Drain_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Drain' | |||
| 119 | + | type MockNoticeStore_Drain_Call struct { | |||
| 120 | + | *mock.Call | |||
| 121 | + | } | |||
| 122 | + | ||||
| 123 | + | // Drain is a helper method to define mock.On call | |||
| 124 | + | func (_e *MockNoticeStore_Expecter) Drain() *MockNoticeStore_Drain_Call { | |||
| 125 | + | return &MockNoticeStore_Drain_Call{Call: _e.mock.On("Drain")} | |||
| 126 | + | } | |||
| 127 | + | ||||
| 128 | + | func (_c *MockNoticeStore_Drain_Call) Run(run func()) *MockNoticeStore_Drain_Call { | |||
| 129 | + | _c.Call.Run(func(args mock.Arguments) { | |||
| 130 | + | run() | |||
| 131 | + | }) | |||
| 132 | + | return _c | |||
| 133 | + | } | |||
| 134 | + | ||||
| 135 | + | func (_c *MockNoticeStore_Drain_Call) Return(notifications []domain.Notification, err error) *MockNoticeStore_Drain_Call { | |||
| 136 | + | _c.Call.Return(notifications, err) | |||
| 137 | + | return _c | |||
| 138 | + | } | |||
| 139 | + | ||||
| 140 | + | func (_c *MockNoticeStore_Drain_Call) RunAndReturn(run func() ([]domain.Notification, error)) *MockNoticeStore_Drain_Call { | |||
| 141 | + | _c.Call.Return(run) | |||
| 142 | + | return _c | |||
| 143 | + | } | |||
A internal/ports/notice.go +10 -0
| @@ -0,0 +1,10 @@ | |||
| 1 | + | package ports | |
| 2 | + | ||
| 3 | + | import "github.com/bnema/vev/internal/domain" | |
| 4 | + | ||
| 5 | + | // NoticeStore persists undeliverable notices across daemon restarts. | |
| 6 | + | type NoticeStore interface { | |
| 7 | + | Append(n domain.Notification) error | |
| 8 | + | // Drain returns all stored notices and truncates the store. | |
| 9 | + | Drain() ([]domain.Notification, error) | |
| 10 | + | } | |
| @@ -0,0 +1,10 @@ | |||||
| 1 | + | package ports | |||
| 2 | + | ||||
| 3 | + | import "github.com/bnema/vev/internal/domain" | |||
| 4 | + | ||||
| 5 | + | // NoticeStore persists undeliverable notices across daemon restarts. | |||
| 6 | + | type NoticeStore interface { | |||
| 7 | + | Append(n domain.Notification) error | |||
| 8 | + | // Drain returns all stored notices and truncates the store. | |||
| 9 | + | Drain() ([]domain.Notification, error) | |||
| 10 | + | } | |||
M internal/usecase/daemon/daemon.go +7 -0
| @@ -114,6 +114,7 @@ type Daemon struct { | |||
| 114 | 114 | persistEnabled bool | |
| 115 | 115 | snaps ports.SnapshotStore | |
| 116 | 116 | snapsEnabled bool | |
| 117 | + | noticeStore ports.NoticeStore | |
| 117 | 118 | snapshotMarshal func(snapcodec.Session) ([]byte, error) | |
| 118 | 119 | snapshotJobs chan *snapshotCapture | |
| 119 | 120 | snapshotWorkerMu sync.Mutex | |
| @@ -240,6 +241,12 @@ func WithSnapshotStore(store ports.SnapshotStore) Option { | |||
| 240 | 241 | } | |
| 241 | 242 | } | |
| 242 | 243 | ||
| 244 | + | // WithNoticeStore enables persisting undeliverable notices across daemon | |
| 245 | + | // restarts. A nil store keeps the daemon in no-op notice-persistence mode. | |
| 246 | + | func WithNoticeStore(store ports.NoticeStore) Option { | |
| 247 | + | return func(d *Daemon) { d.noticeStore = store } | |
| 248 | + | } | |
| 249 | + | ||
| 243 | 250 | // WithCwdReader overrides the process cwd reader used for persistence tests. | |
| 244 | 251 | func WithCwdReader(fn func(int) (string, error)) Option { | |
| 245 | 252 | return func(d *Daemon) { | |
| @@ -114,6 +114,7 @@ type Daemon struct { | |||||
| 114 | persistEnabled bool | 114 | persistEnabled bool | ||
| 115 | snaps ports.SnapshotStore | 115 | snaps ports.SnapshotStore | ||
| 116 | snapsEnabled bool | 116 | snapsEnabled bool | ||
| 117 | + | noticeStore ports.NoticeStore | |||
| 117 | snapshotMarshal func(snapcodec.Session) ([]byte, error) | 118 | snapshotMarshal func(snapcodec.Session) ([]byte, error) | ||
| 118 | snapshotJobs chan *snapshotCapture | 119 | snapshotJobs chan *snapshotCapture | ||
| 119 | snapshotWorkerMu sync.Mutex | 120 | snapshotWorkerMu sync.Mutex | ||
| @@ -240,6 +241,12 @@ func WithSnapshotStore(store ports.SnapshotStore) Option { | |||||
| 240 | } | 241 | } | ||
| 241 | } | 242 | } | ||
| 242 | 243 | ||||
| 244 | + | // WithNoticeStore enables persisting undeliverable notices across daemon | |||
| 245 | + | // restarts. A nil store keeps the daemon in no-op notice-persistence mode. | |||
| 246 | + | func WithNoticeStore(store ports.NoticeStore) Option { | |||
| 247 | + | return func(d *Daemon) { d.noticeStore = store } | |||
| 248 | + | } | |||
| 249 | + | ||||
| 243 | // WithCwdReader overrides the process cwd reader used for persistence tests. | 250 | // WithCwdReader overrides the process cwd reader used for persistence tests. | ||
| 244 | func WithCwdReader(fn func(int) (string, error)) Option { | 251 | func WithCwdReader(fn func(int) (string, error)) Option { | ||
| 245 | return func(d *Daemon) { | 252 | return func(d *Daemon) { | ||
A internal/usecase/daemon/notice_persist_test.go +139 -0
| @@ -0,0 +1,139 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "fmt" | |
| 5 | + | "sync/atomic" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/stretchr/testify/mock" | |
| 9 | + | "github.com/stretchr/testify/require" | |
| 10 | + | ||
| 11 | + | "github.com/bnema/vev/internal/domain" | |
| 12 | + | "github.com/bnema/vev/internal/ports" | |
| 13 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 14 | + | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 15 | + | ) | |
| 16 | + | ||
| 17 | + | // TestKillSessionServerShutdownSaturatedAppendsNotice proves that when the | |
| 18 | + | // daemon is shutting down and the snapshot worker is saturated — so there is | |
| 19 | + | // no way to retain the session's terminal state and no attached client to | |
| 20 | + | // toast, because the daemon is dying — the failure is handed to the | |
| 21 | + | // NoticeStore instead of being silently lost. | |
| 22 | + | func TestKillSessionServerShutdownSaturatedAppendsNotice(t *testing.T) { | |
| 23 | + | store := discardSnapshotStore{} | |
| 24 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 25 | + | WithSnapshotStore(store)(d) | |
| 26 | + | startSnapshotEncodeWorker(t, d) | |
| 27 | + | ||
| 28 | + | entered := make(chan struct{}) | |
| 29 | + | release := make(chan struct{}) | |
| 30 | + | defer close(release) | |
| 31 | + | var calls atomic.Int32 | |
| 32 | + | d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) { | |
| 33 | + | if calls.Add(1) == 1 { | |
| 34 | + | close(entered) | |
| 35 | + | <-release | |
| 36 | + | } | |
| 37 | + | return snapcodec.Marshal(s) | |
| 38 | + | } | |
| 39 | + | ||
| 40 | + | active := newSnapshotTestSession(t, "active", false, "/work") | |
| 41 | + | queued := newSnapshotTestSession(t, "queued", false, "/work") | |
| 42 | + | markSnapshotDirty(active) | |
| 43 | + | markSnapshotDirty(queued) | |
| 44 | + | require.True(t, d.scheduleSnapshot(active)) | |
| 45 | + | <-entered // the worker is now parked inside marshal for active | |
| 46 | + | require.True(t, d.scheduleSnapshot(queued), "one capture waits in the bounded queue") | |
| 47 | + | for i := range snapshotFinalQueueCapacity { | |
| 48 | + | s := newSnapshotTestSession(t, fmt.Sprintf("terminal-%d", i), false, "/work") | |
| 49 | + | markSnapshotDirty(s) | |
| 50 | + | require.True(t, d.scheduleFinalSnapshot(s), "fill the terminal-retention fallback to capacity") | |
| 51 | + | } | |
| 52 | + | ||
| 53 | + | target := newSnapshotTestSession(t, "target", false, "/work") | |
| 54 | + | targetPTY := target.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | |
| 55 | + | targetPTY.EXPECT().Close().Return(nil).Maybe() | |
| 56 | + | d.mu.Lock() | |
| 57 | + | d.sessions[target.id] = target | |
| 58 | + | d.mu.Unlock() | |
| 59 | + | markSnapshotDirty(target) | |
| 60 | + | ||
| 61 | + | notices := portsmocks.NewMockNoticeStore(t) | |
| 62 | + | var appended domain.Notification | |
| 63 | + | notices.EXPECT().Append(mock.Anything).RunAndReturn(func(n domain.Notification) error { | |
| 64 | + | appended = n | |
| 65 | + | return nil | |
| 66 | + | }).Once() | |
| 67 | + | d.noticeStore = notices | |
| 68 | + | ||
| 69 | + | err := d.killSession(target, ports.ReasonServerShutdown, false) | |
| 70 | + | require.ErrorContains(t, err, "snapshot worker unavailable or saturated") | |
| 71 | + | ||
| 72 | + | d.mu.Lock() | |
| 73 | + | _, stillPresent := d.sessions[target.id] | |
| 74 | + | d.mu.Unlock() | |
| 75 | + | require.False(t, stillPresent, "the session must be torn down even though its state could not be saved") | |
| 76 | + | ||
| 77 | + | require.Equal(t, domain.NoticeSnapshotWrite, appended.Code) | |
| 78 | + | require.Equal(t, domain.NoticeError, appended.Severity) | |
| 79 | + | require.Equal(t, "session target shut down without saving terminal state", appended.Message) | |
| 80 | + | require.NotEmpty(t, appended.Details) | |
| 81 | + | } | |
| 82 | + | ||
| 83 | + | // TestKillSessionServerShutdownNotSaturatedSkipsAppend proves the notice store | |
| 84 | + | // is left untouched when the final snapshot is retained normally. | |
| 85 | + | func TestKillSessionServerShutdownNotSaturatedSkipsAppend(t *testing.T) { | |
| 86 | + | store := discardSnapshotStore{} | |
| 87 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 88 | + | WithSnapshotStore(store)(d) | |
| 89 | + | startSnapshotEncodeWorker(t, d) | |
| 90 | + | ||
| 91 | + | notices := portsmocks.NewMockNoticeStore(t) // no .EXPECT(): Append must not be called | |
| 92 | + | d.noticeStore = notices | |
| 93 | + | ||
| 94 | + | target := newSnapshotTestSession(t, "target", false, "/work") | |
| 95 | + | targetPTY := target.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | |
| 96 | + | targetPTY.EXPECT().Close().Return(nil).Maybe() | |
| 97 | + | d.mu.Lock() | |
| 98 | + | d.sessions[target.id] = target | |
| 99 | + | d.mu.Unlock() | |
| 100 | + | markSnapshotDirty(target) | |
| 101 | + | ||
| 102 | + | err := d.killSession(target, ports.ReasonServerShutdown, false) | |
| 103 | + | require.NoError(t, err) | |
| 104 | + | } | |
| 105 | + | ||
| 106 | + | // TestStartupDrainRecordsAndToastsPendingNotices proves notices persisted by a | |
| 107 | + | // previous daemon are recorded to history and toasted to the first client that | |
| 108 | + | // attaches after this daemon starts. | |
| 109 | + | func TestStartupDrainRecordsAndToastsPendingNotices(t *testing.T) { | |
| 110 | + | drained := []domain.Notification{ | |
| 111 | + | {Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session a shut down without saving terminal state"}, | |
| 112 | + | {Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session b shut down without saving terminal state"}, | |
| 113 | + | } | |
| 114 | + | notices := portsmocks.NewMockNoticeStore(t) | |
| 115 | + | notices.EXPECT().Drain().Return(drained, nil).Once() | |
| 116 | + | ||
| 117 | + | p, releasePTY := newBlockingPTY(t) | |
| 118 | + | defer releasePTY() | |
| 119 | + | d := newTestDaemon(t, newFactory(t, p), stubClock{}) | |
| 120 | + | d.noticeStore = notices | |
| 121 | + | ||
| 122 | + | d.restoreSnapshots(t.Context()) | |
| 123 | + | ||
| 124 | + | require.Len(t, d.notices.history(), 2, "each drained notice is recorded individually") | |
| 125 | + | ||
| 126 | + | tr, _ := newCapturingTransport(t) | |
| 127 | + | sess, ac, err := d.route(ports.Hello{ | |
| 128 | + | Version: ports.ProtocolVersion, | |
| 129 | + | Intent: ports.IntentEphemeral, | |
| 130 | + | Size: domain.Size{Cols: 80, Rows: 24}, | |
| 131 | + | }, tr) | |
| 132 | + | require.NoError(t, err) | |
| 133 | + | d.firstPaint(sess, ac, ac.size) | |
| 134 | + | ||
| 135 | + | toasts := awaitToastCount(t, ac, 1) | |
| 136 | + | require.Equal(t, domain.NoticeSnapshotWrite, toasts[0].Code) | |
| 137 | + | require.Equal(t, 2, toasts[0].Count, "two same-code drained notices dedup into one toast counted twice") | |
| 138 | + | require.Empty(t, d.notices.drainPending(), "firstPaint must consume the queue") | |
| 139 | + | } | |
| @@ -0,0 +1,139 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "fmt" | |||
| 5 | + | "sync/atomic" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/stretchr/testify/mock" | |||
| 9 | + | "github.com/stretchr/testify/require" | |||
| 10 | + | ||||
| 11 | + | "github.com/bnema/vev/internal/domain" | |||
| 12 | + | "github.com/bnema/vev/internal/ports" | |||
| 13 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |||
| 14 | + | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |||
| 15 | + | ) | |||
| 16 | + | ||||
| 17 | + | // TestKillSessionServerShutdownSaturatedAppendsNotice proves that when the | |||
| 18 | + | // daemon is shutting down and the snapshot worker is saturated — so there is | |||
| 19 | + | // no way to retain the session's terminal state and no attached client to | |||
| 20 | + | // toast, because the daemon is dying — the failure is handed to the | |||
| 21 | + | // NoticeStore instead of being silently lost. | |||
| 22 | + | func TestKillSessionServerShutdownSaturatedAppendsNotice(t *testing.T) { | |||
| 23 | + | store := discardSnapshotStore{} | |||
| 24 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 25 | + | WithSnapshotStore(store)(d) | |||
| 26 | + | startSnapshotEncodeWorker(t, d) | |||
| 27 | + | ||||
| 28 | + | entered := make(chan struct{}) | |||
| 29 | + | release := make(chan struct{}) | |||
| 30 | + | defer close(release) | |||
| 31 | + | var calls atomic.Int32 | |||
| 32 | + | d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) { | |||
| 33 | + | if calls.Add(1) == 1 { | |||
| 34 | + | close(entered) | |||
| 35 | + | <-release | |||
| 36 | + | } | |||
| 37 | + | return snapcodec.Marshal(s) | |||
| 38 | + | } | |||
| 39 | + | ||||
| 40 | + | active := newSnapshotTestSession(t, "active", false, "/work") | |||
| 41 | + | queued := newSnapshotTestSession(t, "queued", false, "/work") | |||
| 42 | + | markSnapshotDirty(active) | |||
| 43 | + | markSnapshotDirty(queued) | |||
| 44 | + | require.True(t, d.scheduleSnapshot(active)) | |||
| 45 | + | <-entered // the worker is now parked inside marshal for active | |||
| 46 | + | require.True(t, d.scheduleSnapshot(queued), "one capture waits in the bounded queue") | |||
| 47 | + | for i := range snapshotFinalQueueCapacity { | |||
| 48 | + | s := newSnapshotTestSession(t, fmt.Sprintf("terminal-%d", i), false, "/work") | |||
| 49 | + | markSnapshotDirty(s) | |||
| 50 | + | require.True(t, d.scheduleFinalSnapshot(s), "fill the terminal-retention fallback to capacity") | |||
| 51 | + | } | |||
| 52 | + | ||||
| 53 | + | target := newSnapshotTestSession(t, "target", false, "/work") | |||
| 54 | + | targetPTY := target.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | |||
| 55 | + | targetPTY.EXPECT().Close().Return(nil).Maybe() | |||
| 56 | + | d.mu.Lock() | |||
| 57 | + | d.sessions[target.id] = target | |||
| 58 | + | d.mu.Unlock() | |||
| 59 | + | markSnapshotDirty(target) | |||
| 60 | + | ||||
| 61 | + | notices := portsmocks.NewMockNoticeStore(t) | |||
| 62 | + | var appended domain.Notification | |||
| 63 | + | notices.EXPECT().Append(mock.Anything).RunAndReturn(func(n domain.Notification) error { | |||
| 64 | + | appended = n | |||
| 65 | + | return nil | |||
| 66 | + | }).Once() | |||
| 67 | + | d.noticeStore = notices | |||
| 68 | + | ||||
| 69 | + | err := d.killSession(target, ports.ReasonServerShutdown, false) | |||
| 70 | + | require.ErrorContains(t, err, "snapshot worker unavailable or saturated") | |||
| 71 | + | ||||
| 72 | + | d.mu.Lock() | |||
| 73 | + | _, stillPresent := d.sessions[target.id] | |||
| 74 | + | d.mu.Unlock() | |||
| 75 | + | require.False(t, stillPresent, "the session must be torn down even though its state could not be saved") | |||
| 76 | + | ||||
| 77 | + | require.Equal(t, domain.NoticeSnapshotWrite, appended.Code) | |||
| 78 | + | require.Equal(t, domain.NoticeError, appended.Severity) | |||
| 79 | + | require.Equal(t, "session target shut down without saving terminal state", appended.Message) | |||
| 80 | + | require.NotEmpty(t, appended.Details) | |||
| 81 | + | } | |||
| 82 | + | ||||
| 83 | + | // TestKillSessionServerShutdownNotSaturatedSkipsAppend proves the notice store | |||
| 84 | + | // is left untouched when the final snapshot is retained normally. | |||
| 85 | + | func TestKillSessionServerShutdownNotSaturatedSkipsAppend(t *testing.T) { | |||
| 86 | + | store := discardSnapshotStore{} | |||
| 87 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 88 | + | WithSnapshotStore(store)(d) | |||
| 89 | + | startSnapshotEncodeWorker(t, d) | |||
| 90 | + | ||||
| 91 | + | notices := portsmocks.NewMockNoticeStore(t) // no .EXPECT(): Append must not be called | |||
| 92 | + | d.noticeStore = notices | |||
| 93 | + | ||||
| 94 | + | target := newSnapshotTestSession(t, "target", false, "/work") | |||
| 95 | + | targetPTY := target.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | |||
| 96 | + | targetPTY.EXPECT().Close().Return(nil).Maybe() | |||
| 97 | + | d.mu.Lock() | |||
| 98 | + | d.sessions[target.id] = target | |||
| 99 | + | d.mu.Unlock() | |||
| 100 | + | markSnapshotDirty(target) | |||
| 101 | + | ||||
| 102 | + | err := d.killSession(target, ports.ReasonServerShutdown, false) | |||
| 103 | + | require.NoError(t, err) | |||
| 104 | + | } | |||
| 105 | + | ||||
| 106 | + | // TestStartupDrainRecordsAndToastsPendingNotices proves notices persisted by a | |||
| 107 | + | // previous daemon are recorded to history and toasted to the first client that | |||
| 108 | + | // attaches after this daemon starts. | |||
| 109 | + | func TestStartupDrainRecordsAndToastsPendingNotices(t *testing.T) { | |||
| 110 | + | drained := []domain.Notification{ | |||
| 111 | + | {Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session a shut down without saving terminal state"}, | |||
| 112 | + | {Code: domain.NoticeSnapshotWrite, Severity: domain.NoticeError, Message: "session b shut down without saving terminal state"}, | |||
| 113 | + | } | |||
| 114 | + | notices := portsmocks.NewMockNoticeStore(t) | |||
| 115 | + | notices.EXPECT().Drain().Return(drained, nil).Once() | |||
| 116 | + | ||||
| 117 | + | p, releasePTY := newBlockingPTY(t) | |||
| 118 | + | defer releasePTY() | |||
| 119 | + | d := newTestDaemon(t, newFactory(t, p), stubClock{}) | |||
| 120 | + | d.noticeStore = notices | |||
| 121 | + | ||||
| 122 | + | d.restoreSnapshots(t.Context()) | |||
| 123 | + | ||||
| 124 | + | require.Len(t, d.notices.history(), 2, "each drained notice is recorded individually") | |||
| 125 | + | ||||
| 126 | + | tr, _ := newCapturingTransport(t) | |||
| 127 | + | sess, ac, err := d.route(ports.Hello{ | |||
| 128 | + | Version: ports.ProtocolVersion, | |||
| 129 | + | Intent: ports.IntentEphemeral, | |||
| 130 | + | Size: domain.Size{Cols: 80, Rows: 24}, | |||
| 131 | + | }, tr) | |||
| 132 | + | require.NoError(t, err) | |||
| 133 | + | d.firstPaint(sess, ac, ac.size) | |||
| 134 | + | ||||
| 135 | + | toasts := awaitToastCount(t, ac, 1) | |||
| 136 | + | require.Equal(t, domain.NoticeSnapshotWrite, toasts[0].Code) | |||
| 137 | + | require.Equal(t, 2, toasts[0].Count, "two same-code drained notices dedup into one toast counted twice") | |||
| 138 | + | require.Empty(t, d.notices.drainPending(), "firstPaint must consume the queue") | |||
| 139 | + | } | |||
M internal/usecase/daemon/session.go +13 -0
| @@ -723,6 +723,19 @@ func (d *Daemon) killSession(sess *session, reason uint8, purge bool) error { | |||
| 723 | 723 | if reason != ports.ReasonServerShutdown { | |
| 724 | 724 | return terminalSnapshotErr | |
| 725 | 725 | } | |
| 726 | + | // The daemon is dying: there is no attached client left to toast, so | |
| 727 | + | // persist the failure for the next startup to surface instead. | |
| 728 | + | if ns := d.noticeStore; ns != nil { | |
| 729 | + | if err := ns.Append(domain.Notification{ | |
| 730 | + | Code: domain.NoticeSnapshotWrite, | |
| 731 | + | Severity: domain.NoticeError, | |
| 732 | + | Message: "session " + name + " shut down without saving terminal state", | |
| 733 | + | Details: noticeDetails(terminalSnapshotErr), | |
| 734 | + | Time: d.clock.Now(), | |
| 735 | + | }); err != nil { | |
| 736 | + | d.log.Warn("persisting shutdown notice failed", "err", err, "session", name) | |
| 737 | + | } | |
| 738 | + | } | |
| 726 | 739 | } | |
| 727 | 740 | } | |
| 728 | 741 | } | |
| @@ -723,6 +723,19 @@ func (d *Daemon) killSession(sess *session, reason uint8, purge bool) error { | |||||
| 723 | if reason != ports.ReasonServerShutdown { | 723 | if reason != ports.ReasonServerShutdown { | ||
| 724 | return terminalSnapshotErr | 724 | return terminalSnapshotErr | ||
| 725 | } | 725 | } | ||
| 726 | + | // The daemon is dying: there is no attached client left to toast, so | |||
| 727 | + | // persist the failure for the next startup to surface instead. | |||
| 728 | + | if ns := d.noticeStore; ns != nil { | |||
| 729 | + | if err := ns.Append(domain.Notification{ | |||
| 730 | + | Code: domain.NoticeSnapshotWrite, | |||
| 731 | + | Severity: domain.NoticeError, | |||
| 732 | + | Message: "session " + name + " shut down without saving terminal state", | |||
| 733 | + | Details: noticeDetails(terminalSnapshotErr), | |||
| 734 | + | Time: d.clock.Now(), | |||
| 735 | + | }); err != nil { | |||
| 736 | + | d.log.Warn("persisting shutdown notice failed", "err", err, "session", name) | |||
| 737 | + | } | |||
| 738 | + | } | |||
| 726 | } | 739 | } | ||
| 727 | } | 740 | } | ||
| 728 | } | 741 | } | ||
M internal/usecase/daemon/snapshot.go +11 -1
| @@ -63,7 +63,17 @@ func (d *Daemon) closeRestoreDone() { | |||
| 63 | 63 | ||
| 64 | 64 | func (d *Daemon) restoreSnapshots(ctx context.Context) { | |
| 65 | 65 | defer d.closeRestoreDone() | |
| 66 | - | if d == nil || !d.snapsEnabled || d.snaps == nil { | |
| 66 | + | if d == nil { | |
| 67 | + | return | |
| 68 | + | } | |
| 69 | + | if ns := d.noticeStore; ns != nil { | |
| 70 | + | drained, _ := ns.Drain() | |
| 71 | + | for _, n := range drained { | |
| 72 | + | d.notices.record(n) | |
| 73 | + | d.notices.queueGlobal(n) | |
| 74 | + | } | |
| 75 | + | } | |
| 76 | + | if !d.snapsEnabled || d.snaps == nil { | |
| 67 | 77 | return | |
| 68 | 78 | } | |
| 69 | 79 | blobs, err := d.snaps.Load() | |
| @@ -63,7 +63,17 @@ func (d *Daemon) closeRestoreDone() { | |||||
| 63 | 63 | ||||
| 64 | func (d *Daemon) restoreSnapshots(ctx context.Context) { | 64 | func (d *Daemon) restoreSnapshots(ctx context.Context) { | ||
| 65 | defer d.closeRestoreDone() | 65 | defer d.closeRestoreDone() | ||
| 66 | - | if d == nil || !d.snapsEnabled || d.snaps == nil { | 66 | + | if d == nil { |
| 67 | + | return | |||
| 68 | + | } | |||
| 69 | + | if ns := d.noticeStore; ns != nil { | |||
| 70 | + | drained, _ := ns.Drain() | |||
| 71 | + | for _, n := range drained { | |||
| 72 | + | d.notices.record(n) | |||
| 73 | + | d.notices.queueGlobal(n) | |||
| 74 | + | } | |||
| 75 | + | } | |||
| 76 | + | if !d.snapsEnabled || d.snaps == nil { | |||
| 67 | return | 77 | return | ||
| 68 | } | 78 | } | ||
| 69 | blobs, err := d.snaps.Load() | 79 | blobs, err := d.snaps.Load() | ||