1package recovery
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sync"
8 "testing"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/stretchr/testify/require"
13)
14
15type checkpointCatalogue struct {
16 mu sync.Mutex
17 record domain.CatalogueRecord
18 replaceErr error
19 replaces int
20 events *[]string
21}
22
23func (c *checkpointCatalogue) Records() ([]domain.CatalogueRecord, error) {
24 c.mu.Lock()
25 defer c.mu.Unlock()
26 return []domain.CatalogueRecord{c.record}, nil
27}
28func (c *checkpointCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
29 c.mu.Lock()
30 defer c.mu.Unlock()
31 return c.record, c.record.Name == name, nil
32}
33func (c *checkpointCatalogue) Create(domain.CatalogueRecord) error { return nil }
34func (c *checkpointCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
35 return nil
36}
37func (c *checkpointCatalogue) Sync() error { return nil }
38func (c *checkpointCatalogue) Replace(name string, record domain.CatalogueRecord) error {
39 c.mu.Lock()
40 defer c.mu.Unlock()
41 c.replaces++
42 if c.events != nil {
43 *c.events = append(*c.events, "catalogue")
44 }
45 if c.replaceErr != nil {
46 return c.replaceErr
47 }
48 if name != c.record.Name || record.Name != name {
49 return errors.New("catalogue key mismatch")
50 }
51 c.record = record
52 return nil
53}
54func (c *checkpointCatalogue) Rename(string, domain.CatalogueRecord) error { return nil }
55func (c *checkpointCatalogue) Delete(string) error { return nil }
56func (c *checkpointCatalogue) Close() error { return nil }
57
58type checkpointRepository struct {
59 ports.SnapshotRepository
60 publications []ports.SnapshotPublication
61 generations map[uint64]ports.SnapshotGeneration
62 loadErrors map[uint64]error
63 loads []uint64
64 repairs []domain.CheckpointRef
65 repairErr error
66 events *[]string
67}
68
69func (r *checkpointRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
70 if r.events != nil {
71 *r.events = append(*r.events, "repository")
72 }
73 r.publications = append(r.publications, publication)
74 return nil
75}
76
77func (r *checkpointRepository) LoadCheckpoint(_ context.Context, _ domain.IncarnationID, _ string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
78 r.loads = append(r.loads, ref.Generation)
79 if err := r.loadErrors[ref.Generation]; err != nil {
80 return ports.SnapshotGeneration{}, err
81 }
82 generation, ok := r.generations[ref.Generation]
83 if !ok {
84 return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
85 }
86 return generation, nil
87}
88
89func (r *checkpointRepository) ReconcileCheckpoint(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
90 r.repairs = append(r.repairs, ref)
91 return r.repairErr
92}
93
94func checkpointRecord() domain.CatalogueRecord {
95 return domain.CatalogueRecord{
96 Name: "work",
97 IncarnationID: domain.IncarnationID{1},
98 }
99}
100
101func checkpointPublication(record domain.CatalogueRecord, generation uint64) ports.SnapshotPublication {
102 return ports.SnapshotPublication{
103 IncarnationID: record.IncarnationID,
104 Name: record.Name,
105 Generation: generation,
106 ParentCheckpoint: record.Committed,
107 Manifest: []byte{byte(generation), 0xa5},
108 }
109}
110
111func runCheckpointCommit(t *testing.T, count int) (domain.CatalogueRecord, *checkpointCatalogue, *checkpointRepository) {
112 t.Helper()
113 events := make([]string, 0, count*2)
114 catalogue := &checkpointCatalogue{record: checkpointRecord(), events: &events}
115 repository := &checkpointRepository{events: &events}
116 coordinator := NewCoordinator(catalogue, repository, nil)
117 for generation := uint64(1); generation <= uint64(count); generation++ {
118 var err error
119 catalogue.record, err = coordinator.PublishCheckpoint(context.Background(), "work", checkpointPublication(catalogue.record, generation))
120 require.NoError(t, err)
121 }
122 for i := range count {
123 require.Equal(t, []string{"repository", "catalogue"}, events[i*2:i*2+2])
124 }
125 return catalogue.record, catalogue, repository
126}
127
128func TestCheckpointCommits(t *testing.T) {
129 for _, tc := range []struct {
130 name string
131 count int
132 wantCommitted uint64
133 }{
134 {name: "first", count: 1, wantCommitted: 1},
135 {name: "second", count: 2, wantCommitted: 2},
136 {name: "third", count: 3, wantCommitted: 3},
137 {name: "fourth", count: 4, wantCommitted: 4},
138 } {
139 t.Run(tc.name, func(t *testing.T) {
140 record, catalogue, _ := runCheckpointCommit(t, tc.count)
141 require.Equal(t, tc.wantCommitted, record.Committed.Generation)
142 require.Empty(t, record.DegradedReason)
143 require.Equal(t, tc.count, catalogue.replaces)
144 })
145 }
146}
147
148func TestCheckpointRejectsNonIncreasingGeneration(t *testing.T) {
149 prior := checkpointRecord()
150 priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
151 prior.Committed = &priorRef
152
153 for _, generation := range []uint64{7, 6} {
154 t.Run(fmt.Sprintf("generation-%d", generation), func(t *testing.T) {
155 catalogue := &checkpointCatalogue{record: prior}
156 repository := &checkpointRepository{}
157 coordinator := NewCoordinator(catalogue, repository, nil)
158
159 _, err := coordinator.PublishCheckpoint(context.Background(), prior.Name, checkpointPublication(prior, generation))
160
161 require.ErrorIs(t, err, ErrCheckpointConflict)
162 require.Empty(t, repository.publications)
163 require.Zero(t, catalogue.replaces)
164 })
165 }
166}
167
168func TestCheckpointCommitPublishOrphan(t *testing.T) {
169 prior := checkpointRecord()
170 priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
171 prior.Committed = &priorRef
172 cause := errors.New("catalogue sync failed")
173 catalogue := &checkpointCatalogue{record: prior, replaceErr: cause}
174 repository := &checkpointRepository{}
175 coordinator := NewCoordinator(catalogue, repository, nil)
176 publication := checkpointPublication(prior, 8)
177
178 _, err := coordinator.PublishCheckpoint(context.Background(), prior.Name, publication)
179 require.ErrorIs(t, err, cause)
180 require.Equal(t, prior, catalogue.record, "failed catalogue commit must preserve the authoritative record")
181 require.Len(t, repository.publications, 1, "the durable publication remains a forward orphan")
182 require.Equal(t, publication, repository.publications[0])
183}