checkpoint.go RawHistoryFolder

1package recovery
2
3import (
4	"context"
5	"crypto/sha256"
6	"errors"
7	"fmt"
8
9	"github.com/bnema/vev/internal/domain"
10	"github.com/bnema/vev/internal/ports"
11)
12
13var (
14	ErrCheckpointRecordNotFound = errors.New("recovery: checkpoint catalogue record not found")
15	ErrCheckpointConflict       = errors.New("recovery: checkpoint publication conflicts with catalogue")
16)
17
18// PublishCheckpoint publishes repository data before atomically making its
19// reference authoritative in the catalogue. A catalogue failure deliberately
20// leaves the durable publication as a same-incarnation forward orphan.
21func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, publication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
22	if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
23		return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
24	}
25	if err := ctx.Err(); err != nil {
26		return domain.CatalogueRecord{}, err
27	}
28	c.mutationMu.Lock()
29	defer c.mutationMu.Unlock()
30	unlock := c.locks.Lock([]string{name})
31	defer unlock()
32
33	record, ok, err := c.catalogue.Record(name)
34	if err != nil {
35		return domain.CatalogueRecord{}, err
36	}
37	if !ok {
38		return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
39	}
40	currentGeneration := uint64(0)
41	if record.Committed != nil {
42		currentGeneration = record.Committed.Generation
43	}
44	if publication.Name != name || publication.Name != record.Name || publication.IncarnationID != record.IncarnationID ||
45		publication.Generation <= currentGeneration || len(publication.Manifest) == 0 ||
46		!equalCheckpointRefs(publication.ParentCheckpoint, record.Committed) {
47		return domain.CatalogueRecord{}, ErrCheckpointConflict
48	}
49	if err := c.repository.Publish(ctx, publication); err != nil {
50		return domain.CatalogueRecord{}, err
51	}
52
53	ref := domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
54	next := record
55	next.Committed = copyCheckpointRef(&ref)
56	next.DegradedReason = ""
57	if err := next.Validate(); err != nil {
58		return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid checkpoint transition: %w", err)
59	}
60	if err := c.catalogue.Replace(name, next); err != nil {
61		return domain.CatalogueRecord{}, err
62	}
63	return next, nil
64}
65
66func equalCheckpointRefs(a, b *domain.CheckpointRef) bool { return a.Equal(b) }
67
68func copyCheckpointRef(ref *domain.CheckpointRef) *domain.CheckpointRef {
69	if ref == nil {
70		return nil
71	}
72	cloned := *ref
73	return &cloned
74}