1package persist
2
3import (
4 "encoding/binary"
5 "os"
6 "path/filepath"
7 "testing"
8
9 "github.com/stretchr/testify/require"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/protocol"
13 "github.com/bnema/vev/pkg/kv"
14 "github.com/bnema/vev/pkg/safedir"
15)
16
17func TestCatalogue(t *testing.T) {
18 t.Run("record-round-trip", testCatalogueRecordRoundTrip)
19 t.Run("duplicate-incarnations", testCatalogueDuplicateIncarnations)
20 t.Run("open-fails-closed", testCatalogueOpenFailsClosed)
21 t.Run("apply-rename-replace", testCatalogueApplyRenameReplace)
22 t.Run("metadata-update-preserves-authority", testCatalogueMetadataUpdatePreservesAuthority)
23 t.Run("metadata-update-is-deferred", testCatalogueMetadataUpdateIsDeferred)
24 t.Run("open-rejects-malformed", testCatalogueOpenRejectsMalformedValue)
25 t.Run("open-ignores-stray-tmp", testCatalogueOpenIgnoresStrayTmp)
26}
27
28func testCatalogueRecordRoundTrip(t *testing.T) {
29 ref1 := &domain.CheckpointRef{Generation: 9, ManifestDigest: [32]byte{1}}
30 tests := []domain.CatalogueRecord{
31 {Name: "fresh", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp"},
32 {Name: "healthy", IncarnationID: domain.IncarnationID{2}, Committed: ref1},
33 {Name: "broken", IncarnationID: domain.IncarnationID{4}, Committed: ref1, DegradedReason: "invalid manifest"},
34 {Name: "other-fresh", IncarnationID: domain.IncarnationID{5}},
35 }
36 for _, record := range tests {
37 t.Run(record.Name, func(t *testing.T) {
38 encoded, err := encodeRecordValue(record)
39 require.NoError(t, err)
40 got, err := decodeRecordValue(record.Name, encoded)
41 require.NoError(t, err)
42 require.Equal(t, record, got)
43 for prefix := range len(encoded) {
44 _, err := decodeRecordValue(record.Name, encoded[:prefix])
45 require.Error(t, err, "prefix length %d", prefix)
46 }
47 require.Error(t, func() error { _, err := decodeRecordValue(record.Name, append(encoded, 0)); return err }())
48 })
49 }
50}
51
52func TestOpenOrCreateMigratesLegacyRecordsWithoutProtocolReset(t *testing.T) {
53 record := domain.CatalogueRecord{
54 Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
55 CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
56 TabNames: []string{"shell", "logs"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "shell", Name: "shell"}, {StableID: "logs", Name: "logs"}},
57 Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
58 }
59 for _, tt := range []struct {
60 name string
61 format uint16
62 wire uint16
63 }{
64 {name: "v5 older protocol", format: protocolRecordVersion, wire: protocol.Version - 1},
65 {name: "v5 current protocol", format: protocolRecordVersion, wire: protocol.Version},
66 {name: "v5 newer protocol", format: protocolRecordVersion, wire: protocol.Version + 1},
67 {name: "v4", format: protocolLessRecordVersion},
68 } {
69 t.Run(tt.name, func(t *testing.T) {
70 encoded, err := encodeRecordValueForFormat(record, tt.format, tt.wire)
71 require.NoError(t, err)
72 dir := privateDir(t)
73 store, err := kv.Open(StorePath(dir))
74 require.NoError(t, err)
75 require.NoError(t, store.Set([]byte(record.Name), encoded))
76 require.NoError(t, store.Close())
77
78 opened, err := OpenOrCreate(dir)
79 require.NoError(t, err)
80 require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
81 require.True(t, opened.Migration.Performed)
82 require.Equal(t, []uint16{tt.format}, opened.Migration.SourceFormats)
83 require.NoError(t, opened.Catalogue.Close())
84 migrated, err := kv.Open(StorePath(dir))
85 require.NoError(t, err)
86 value, ok := migrated.Get([]byte(record.Name))
87 require.True(t, ok)
88 require.NoError(t, migrated.Close())
89 stored, err := decodeStoredRecordValue(record.Name, value)
90 require.NoError(t, err)
91 require.Equal(t, catalogueRecordVersion, stored.formatVersion)
92 })
93 }
94}
95
96func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {
97 encoded, err := encodeRecordValue(validRecord("work", 1))
98 require.NoError(t, err)
99 encoded[checkpointMarkerOffset(t, encoded)] = 99
100
101 _, err = decodeRecordValue("work", encoded)
102 require.ErrorIs(t, err, errMalformedRecord)
103}
104
105func TestDecodeRecordValueRejectsUnsupportedFormat(t *testing.T) {
106 encoded, err := encodeRecordValue(validRecord("work", 1))
107 require.NoError(t, err)
108 binary.BigEndian.PutUint16(encoded[len(catalogueMagic):], catalogueRecordVersion+1)
109
110 _, err = decodeRecordValue("work", encoded)
111 var unsupported *UnsupportedCatalogueRecordFormatError
112 require.ErrorAs(t, err, &unsupported)
113 require.Equal(t, catalogueRecordVersion+1, unsupported.Got)
114}
115
116func checkpointMarkerOffset(t *testing.T, encoded []byte) int {
117 t.Helper()
118 r := valueReader{data: encoded}
119 _, ok := r.take(len(catalogueMagic))
120 require.True(t, ok)
121 version, ok := r.u16()
122 require.True(t, ok)
123 if version == protocolRecordVersion {
124 _, ok = r.u16()
125 require.True(t, ok)
126 }
127 _, ok = r.take(len(domain.IncarnationID{}))
128 require.True(t, ok)
129 _, ok = r.str()
130 require.True(t, ok)
131 _, ok = r.take(8 + 8 + 8)
132 require.True(t, ok)
133 count, ok := r.u32()
134 require.True(t, ok)
135 for range count {
136 _, ok = r.str()
137 require.True(t, ok)
138 }
139 return len(encoded) - r.remaining()
140}
141
142func testCatalogueDuplicateIncarnations(t *testing.T) {
143 id := domain.IncarnationID{1}
144 require.Error(t, validateUniqueIncarnations([]domain.CatalogueRecord{
145 {Name: "one", IncarnationID: id},
146 {Name: "two", IncarnationID: id},
147 }))
148}
149
150func testCatalogueOpenFailsClosed(t *testing.T) {
151 dir := privateDir(t)
152 _, _, err := openCurrentCatalogue(dir, false)
153 require.Error(t, err)
154
155 require.NoError(t, safedir.EnsurePrivate(dir))
156 raw := []byte("corrupt catalogue")
157 require.NoError(t, os.WriteFile(StorePath(dir), raw, 0o600))
158 _, _, err = openCurrentCatalogue(dir, false)
159 require.ErrorIs(t, err, kv.ErrCorrupt)
160 after, readErr := os.ReadFile(StorePath(dir))
161 require.NoError(t, readErr)
162 require.Equal(t, raw, after)
163}
164
165func testCatalogueApplyRenameReplace(t *testing.T) {
166 dir := privateDir(t)
167 p, records, err := openCurrentCatalogue(dir, true)
168 require.NoError(t, err)
169 require.Empty(t, records)
170 defer func() { require.NoError(t, p.Close()) }()
171
172 one := domain.CatalogueRecord{Name: "one", IncarnationID: domain.IncarnationID{1}}
173 two := domain.CatalogueRecord{Name: "two", IncarnationID: domain.IncarnationID{2}}
174 require.NoError(t, p.Apply(map[string]*domain.CatalogueRecord{"one": &one, "two": &two}))
175 renamed := one
176 renamed.Name = "renamed"
177 require.NoError(t, p.Rename("one", renamed))
178 two.Cwd = "/next"
179 require.NoError(t, p.Replace("two", two))
180
181 got, err := p.LoadCatalogue()
182 require.NoError(t, err)
183 require.ElementsMatch(t, []domain.CatalogueRecord{renamed, two}, got)
184}
185
186func testCatalogueMetadataUpdatePreservesAuthority(t *testing.T) {
187 dir := privateDir(t)
188 p, _, err := openCurrentCatalogue(dir, true)
189 require.NoError(t, err)
190 defer func() { require.NoError(t, p.Close()) }()
191
192 committed := &domain.CheckpointRef{Generation: 9, ManifestDigest: [32]byte{1}}
193 original := domain.CatalogueRecord{
194 Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/old", CreatedAt: 11,
195 UpdatedAt: 12, LastUsedSeq: 13, TabNames: []string{"old"},
196 Committed: committed, DegradedReason: "repair pending",
197 }
198 require.NoError(t, p.Create(original))
199
200 next := domain.CatalogueRecord{Name: "work", IncarnationID: original.IncarnationID, Cwd: "/new", UpdatedAt: 22, LastUsedSeq: 23, TabNames: []string{"editor", "logs"}}
201 update := next.MetadataUpdate()
202 require.NoError(t, p.UpdateMetadata(update))
203 got, ok, err := p.Record("work")
204 require.NoError(t, err)
205 require.True(t, ok)
206 expected := original
207 expected.Cwd, expected.UpdatedAt, expected.LastUsedSeq, expected.TabNames = next.Cwd, next.UpdatedAt, next.LastUsedSeq, next.TabNames
208 require.Equal(t, expected, got)
209
210 lastUsedSeq := uint64(24)
211 require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
212 Name: "work", IncarnationID: original.IncarnationID, LastUsedSeq: &lastUsedSeq,
213 }))
214 expected.LastUsedSeq = lastUsedSeq
215 partiallyUpdated, ok, err := p.Record("work")
216 require.NoError(t, err)
217 require.True(t, ok)
218 require.Equal(t, expected, partiallyUpdated, "a partial metadata update must not clear other mutable or authority-owned fields")
219
220 stale := update
221 stale.IncarnationID = domain.IncarnationID{2}
222 require.Error(t, p.UpdateMetadata(stale))
223 unchanged, ok, err := p.Record("work")
224 require.NoError(t, err)
225 require.True(t, ok)
226 require.Equal(t, expected, unchanged)
227}
228
229func testCatalogueMetadataUpdateIsDeferred(t *testing.T) {
230 dir := privateDir(t)
231 p, _, err := openCurrentCatalogue(dir, true)
232 require.NoError(t, err)
233 defer func() { require.NoError(t, p.Close()) }()
234
235 record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/old"}
236 require.NoError(t, p.Create(record))
237 before, err := os.ReadFile(StorePath(dir))
238 require.NoError(t, err)
239
240 cwd := "/buffered"
241 require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
242 Name: record.Name, IncarnationID: record.IncarnationID, Cwd: &cwd,
243 }))
244 afterBuffered, err := os.ReadFile(StorePath(dir))
245 require.NoError(t, err)
246 require.Equal(t, before, afterBuffered, "metadata must remain buffered before Sync")
247
248 require.NoError(t, p.Sync())
249 afterSync, err := os.ReadFile(StorePath(dir))
250 require.NoError(t, err)
251 require.NotEqual(t, before, afterSync)
252
253 cwd = "/next-identity"
254 require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
255 Name: record.Name, IncarnationID: record.IncarnationID, Cwd: &cwd,
256 }))
257 record.Cwd = cwd
258 record.UpdatedAt = 2
259 require.NoError(t, p.Replace(record.Name, record))
260 afterIdentity, err := os.ReadFile(StorePath(dir))
261 require.NoError(t, err)
262 require.NotEqual(t, afterSync, afterIdentity, "identity writes must sync before returning")
263}
264
265func testCatalogueOpenRejectsMalformedValue(t *testing.T) {
266 dir := privateDir(t)
267 store, err := kv.Open(filepath.Join(dir, filename))
268 require.NoError(t, err)
269 require.NoError(t, store.Set([]byte("work"), []byte("malformed")))
270 require.NoError(t, store.Close())
271 _, err = Open(dir)
272 require.Error(t, err)
273}
274
275func testCatalogueOpenIgnoresStrayTmp(t *testing.T) {
276 dir := privateDir(t)
277 p, _, err := openCurrentCatalogue(dir, true)
278 require.NoError(t, err)
279 want := validRecord("work", 1)
280 require.NoError(t, p.Create(want))
281 require.NoError(t, p.Close())
282 require.NoError(t, os.WriteFile(StorePath(dir)+".tmp", []byte("partial rewrite"), 0o600))
283
284 reopened, records, err := openCurrentCatalogue(dir, false)
285 require.NoError(t, err)
286 require.NoError(t, reopened.Close())
287 require.Equal(t, []domain.CatalogueRecord{want}, records)
288 tmp, err := os.ReadFile(StorePath(dir) + ".tmp")
289 require.NoError(t, err)
290 require.Equal(t, []byte("partial rewrite"), tmp)
291}