Add checkpoints backend selection settings and mirror fan-out · Entire
Add checkpoints backend selection settings and mirror fan-out
c7b6a00→main·Soph·3w ago·7 files·+690 added/-4 removed
Wire the registry to settings-driven backend selection and add independent backend mirroring.
Settings: a new checkpoints.{primary, mirrors} block using a discriminated {type, config} shape, read through a dedicated lenient loader (settings.LoadCheckpointsConfig). The loader is fail-soft by design — a missing file, whole-file JSON syntax error, or unrelated invalid settings all resolve to "no config" so checkpoint construction defaults to git; it errors only when a present checkpoints block is itself invalid. This avoids making Open newly fail on unrelated malformed settings (the strict settings.Load path still surfaces those for normal commands). The field also lives on EntireSettings so the strict loader accepts a checkpoints key.
Open: builds the primary and mirrors through the registry. The primary must be git (attach/resume/push/doctor/cleanup/OPF all assume a git refs.Primary), so a non-git primary is rejected. A git-typed mirror is rejected too: it would share the primary ref topology and double-write the same ref.
fanoutStore: serves all reads from the primary and writes to the primary first, then fans out best-effort to each mirror (failures logged, never surfaced). With no mirrors it returns the primary unwrapped, preserving its concrete type and optional capabilities. When wrapping, it preserves the optional AuthorReader iff the primary implements it, so explain's author fallback keeps working.
Mirrors are write-only and may lag the primary (no cleanup-delete or pre-push OPF fan-out); they must not become a read/sync source without reconciliation.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
Sessions
c9cefdc270e2View transcript
Changes
7
cmd/entire/cli
- checkpoint
- Afanout.go+98
- Afanout_test.go+168
- Mopen.go+57/-4
- Aopen_config_test.go+121
- checkpoint
package checkpoint
import (
"context"
"github.com/entireio/cli/cmd/entire/cli/checkpoint/id"
"github.com/entireio/cli/cmd/entire/cli/logging"
)
// fanoutStore serves all reads from the primary and fans writes out to the
// primary plus zero or more mirror backends. The primary is the source of
// truth: a write fails only if the primary write fails. Mirror writes are
// best-effort — a mirror failure is logged and swallowed so it can never break
// a checkpoint operation. Mirrors are therefore write-only and may legitimately
// lag the primary (they never receive ref-level mutations such as cleanup
// deletes or pre-push OPF re-redaction); they must not be promoted to a read or
// sync source without separate reconciliation.
type fanoutStore struct {
primary PersistentStore
mirrors []Writer
}
// newFanoutStore wraps a primary with mirror write fan-out. With no mirrors it
// returns the primary unchanged, so the common (no-mirror) path keeps the
// concrete store and all of its optional capabilities. When wrapping is needed,
// it preserves the optional AuthorReader capability iff the primary has it.
func newFanoutStore(primary PersistentStore, mirrors []Writer) PersistentStore {
if len(mirrors) == 0 {
return primary
}
base := &fanoutStore{primary: primary, mirrors: mirrors}
if author, ok := primary.(AuthorReader); ok {
return &fanoutStoreWithAuthor{fanoutStore: base, author: author}
}
return base
}
// The read methods are pure delegation to the primary; nolint:wrapcheck because
// re-wrapping the primary's errors here would add noise without context (same
// convention as the contract re-exports in aliases.go).
func (s *fanoutStore) Read(ctx context.Context, checkpointID id.CheckpointID) (*CheckpointSummary, error) {
return s.primary.Read(ctx, checkpointID) //nolint:wrapcheck // pure delegation to primary
}
func (s *fanoutStore) List(ctx context.Context) ([]CheckpointInfo, error) {
return s.primary.List(ctx) //nolint:wrapcheck // pure delegation to primary
}
func (s *fanoutStore) ReadSessionContent(ctx context.Context, checkpointID id.CheckpointID, sessionIndex int) (*SessionContent, error) {
return s.primary.ReadSessionContent(ctx, checkpointID, sessionIndex) //nolint:wrapcheck // pure delegation to primary
}
func (s *fanoutStore) ReadSessionMetadata(ctx context.Context, checkpointID id.CheckpointID, sessionIndex int) (*Metadata, error) {
return s.primary.ReadSessionMetadata(ctx, checkpointID, sessionIndex) //nolint:wrapcheck // pure delegation to primary
}
func (s *fanoutStore) ReadSessionPrompts(ctx context.Context, checkpointID id.CheckpointID, sessionIndex int) (string, error) {
return s.primary.ReadSessionPrompts(ctx, checkpointID, sessionIndex) //nolint:wrapcheck // pure delegation to primary
}
func (s *fanoutStore) ReadSessionMetadataAndPrompts(ctx context.Context, checkpointID id.CheckpointID, sessionIndex int) (*Metadata, string, error) {
return s.primary.ReadSessionMetadataAndPrompts(ctx, checkpointID, sessionIndex) //nolint:wrapcheck // pure delegation to primary
}
// Write applies to the primary first; only on primary success does it fan out to
// each mirror best-effort. A mirror error is logged and dropped.
func (s *fanoutStore) Write(ctx context.Context, req WriteRequest) error {
if err := s.primary.Write(ctx, req); err != nil {
return err //nolint:wrapcheck // primary error is the operation's error, surfaced verbatim
}
for i, mirror := range s.mirrors {
if err := mirror.Write(ctx, req); err != nil {
logging.Warn(ctx, "checkpoint mirror write failed; primary write succeeded",
"mirror_index", i, "error", err.Error())
}
}
return nil
}
// fanoutStoreWithAuthor adds the optional AuthorReader capability when the
// wrapped primary supports it, so callers that type-assert the store to
// AuthorReader (e.g. explain's author fallback) keep working through the wrapper.
type fanoutStoreWithAuthor struct {
*fanoutStore
author AuthorReader
}
func (s *fanoutStoreWithAuthor) GetCheckpointAuthor(ctx context.Context, checkpointID id.CheckpointID) (Author, error) {
return s.author.GetCheckpointAuthor(ctx, checkpointID) //nolint:wrapcheck // pure delegation to primary
}
var (
_ PersistentStore = (*fanoutStore)(nil)
_ PersistentStore = (*fanoutStoreWithAuthor)(nil)
_ AuthorReader = (*fanoutStoreWithAuthor)(nil)
)
Acmd/entire/cli/checkpoint/fanout.go+98
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
package checkpoint
import (
"context"
"errors"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/entireio/cli/cmd/entire/cli/checkpoint/id"
)
// fakePrimary is a minimal PersistentStore that records writes and reports a
// fixed read result, so tests can assert read delegation and write fan-out.
type fakePrimary struct {
writes []WriteRequest
writeErr error
listErr error
listCall int
}
func (f *fakePrimary) Read(context.Context, id.CheckpointID) (*CheckpointSummary, error) {
return &CheckpointSummary{}, nil
}
func (f *fakePrimary) List(context.Context) ([]CheckpointInfo, error) {
f.listCall++
if f.listErr != nil {
return nil, f.listErr
}
return []CheckpointInfo{{}}, nil
}
func (f *fakePrimary) ReadSessionContent(context.Context, id.CheckpointID, int) (*SessionContent, error) {
return &SessionContent{}, nil
}
func (f *fakePrimary) ReadSessionMetadata(context.Context, id.CheckpointID, int) (*Metadata, error) {
return &Metadata{}, nil
}
func (f *fakePrimary) ReadSessionPrompts(context.Context, id.CheckpointID, int) (string, error) {
return "", nil
}
func (f *fakePrimary) ReadSessionMetadataAndPrompts(context.Context, id.CheckpointID, int) (*Metadata, string, error) {
return &Metadata{}, "", nil
}
func (f *fakePrimary) Write(_ context.Context, req WriteRequest) error {
if f.writeErr != nil {
return f.writeErr
}
f.writes = append(f.writes, req)
return nil
}
// fakePrimaryWithAuthor adds the optional AuthorReader capability.
type fakePrimaryWithAuthor struct {
*fakePrimary
author Author
authorErr error
}
func (f *fakePrimaryWithAuthor) GetCheckpointAuthor(context.Context, id.CheckpointID) (Author, error) {
return f.author, f.authorErr
}
// fakeMirror records the writes it receives and can be made to fail.
type fakeMirror struct {
writes []WriteRequest
writeErr error
}
func (m *fakeMirror) Write(_ context.Context, req WriteRequest) error {
if m.writeErr != nil {
return m.writeErr
}
m.writes = append(m.writes, req)
return nil
}
func TestFanout_NoMirrorsReturnsPrimaryUnwrapped(t *testing.T) {
t.Parallel()
primary := &fakePrimaryWithAuthor{fakePrimary: &fakePrimary{}, author: Author{Name: "A"}}
store := newFanoutStore(primary, nil)
// With no mirrors the primary is returned as-is — same value, no wrapper.
assert.Same(t, any(primary), any(store))
}
func TestFanout_WriteFansOutToAllMirrors(t *testing.T) {
t.Parallel()
primary := &fakePrimary{}
m1, m2 := &fakeMirror{}, &fakeMirror{}
store := newFanoutStore(primary, []Writer{m1, m2})
req := SessionSummary{CheckpointID: id.CheckpointID("abc123def456")}
require.NoError(t, store.Write(context.Background(), req))
assert.Len(t, primary.writes, 1)
assert.Len(t, m1.writes, 1)
assert.Len(t, m2.writes, 1)
}
func TestFanout_MirrorFailureDoesNotFailWrite(t *testing.T) {
t.Parallel()
primary := &fakePrimary{}
failing := &fakeMirror{writeErr: errors.New("mirror down")}
ok := &fakeMirror{}
store := newFanoutStore(primary, []Writer{failing, ok})
// Primary succeeded, so the operation succeeds even though a mirror failed,
// and later mirrors still receive the write.
require.NoError(t, store.Write(context.Background(), SessionSummary{}))
assert.Len(t, primary.writes, 1)
assert.Len(t, ok.writes, 1)
}
func TestFanout_PrimaryFailureSkipsMirrors(t *testing.T) {
t.Parallel()
primary := &fakePrimary{writeErr: errors.New("primary down")}
mirror := &fakeMirror{}
store := newFanoutStore(primary, []Writer{mirror})
err := store.Write(context.Background(), SessionSummary{})
require.Error(t, err)
assert.Contains(t, err.Error(), "primary down")
// The mirror must not be written when the primary write failed.
assert.Empty(t, mirror.writes)
}
func TestFanout_ReadsDelegateToPrimary(t *testing.T) {
t.Parallel()
primary := &fakePrimary{}
store := newFanoutStore(primary, []Writer{&fakeMirror{}})
_, err := store.List(context.Background())
require.NoError(t, err)
assert.Equal(t, 1, primary.listCall)
}
func TestFanout_PreservesAuthorReaderWhenPrimaryHasIt(t *testing.T) {
t.Parallel()
primary := &fakePrimaryWithAuthor{fakePrimary: &fakePrimary{}, author: Author{Name: "Ada", Email: "ada@example.com"}}
store := newFanoutStore(primary, []Writer{&fakeMirror{}})
author, ok := store.(AuthorReader)
require.True(t, ok, "fan-out wrapper should expose AuthorReader when primary does")
got, err := author.GetCheckpointAuthor(context.Background(), id.CheckpointID("abc123def456"))
require.NoError(t, err)
assert.Equal(t, "Ada", got.Name)
}
func TestFanout_OmitsAuthorReaderWhenPrimaryLacksIt(t *testing.T) {
t.Parallel()
primary := &fakePrimary{} // no GetCheckpointAuthor
store := newFanoutStore(primary, []Writer{&fakeMirror{}})
_, ok := store.(AuthorReader)
assert.False(t, ok, "fan-out wrapper must not advertise AuthorReader when primary lacks it")
}
Acmd/entire/cli/checkpoint/fanout_test.go+168
1
2
3
4
5
6
7
8
9
10
11
12
13
22 unmodified lines
36
37
38
35
36
39
40
41
42
43
44
45
40
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
45
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
1 unmodified line
import (
"context"
"encoding/json"
"fmt"
"github.com/go-git/go-git/v6"
"github.com/entireio/cli/cmd/entire/cli/settings"
)
// OpenOptions configures Open. The zero value uses the default committed-ref
22 unmodified lines
// Open resolves the checkpoint storage topology and constructs the backing
// store(s). It keeps ref resolution, backend selection, and blob-fetcher wiring
// in one place. The primary is built through the backend registry; today it
// always resolves to the git backend, so default behavior is unchanged.
// in one place. The primary is built through the backend registry; with no
// checkpoints config it resolves to the git backend with no mirrors, so default
// behavior is unchanged. When mirrors are configured, the persistent store is a
// fan-out wrapper (reads from primary, best-effort writes to each mirror).
func Open(ctx context.Context, repo *git.Repository, opts OpenOptions) (*Stores, error) {
refs := resolveOpenRefs(ctx, opts)
env := OpenEnv{Repo: repo, BlobFetcher: opts.BlobFetcher, Refs: refs}
primary, err := build(ctx, env, BackendTypeGit, nil)
cfg, err := settings.LoadCheckpointsConfig(ctx)
if err != nil {
return nil, fmt.Errorf("resolve checkpoints config: %w", err)
}
primary, err := buildPrimary(ctx, env, cfg)
if err != nil {
return nil, err
}
mirrors, err := buildMirrors(ctx, env, cfg)
if err != nil {
return nil, err
}
return &Stores{
Persistent: primary,
Persistent: newFanoutStore(primary, mirrors),
ephemeral: newEphemeralStore(repo, refs),
refs: refs,
}, nil
}
// buildPrimary constructs the primary persistent store. The primary must be the
// git backend: attach, resume, push, doctor, cleanup, and OPF all assume a git
// refs.Primary, so a non-git primary is rejected rather than silently
// half-supported.
func buildPrimary(ctx context.Context, env OpenEnv, cfg *settings.CheckpointsConfig) (PersistentStore, error) {
typ, raw := BackendTypeGit, json.RawMessage(nil)
if cfg != nil && cfg.Primary.Type != "" {
typ, raw = cfg.Primary.Type, cfg.Primary.Config
}
if typ != BackendTypeGit {
return nil, fmt.Errorf("checkpoints.primary.type %q is not supported: only %q may be the primary backend", typ, BackendTypeGit)
}
return build(ctx, env, typ, raw)
}
// buildMirrors constructs the mirror writers. A git-typed mirror is rejected: it
// would share the primary ref topology and double-write the same ref, so it is
// never a meaningful independent mirror.
func buildMirrors(ctx context.Context, env OpenEnv, cfg *settings.CheckpointsConfig) ([]Writer, error) {
if cfg == nil || len(cfg.Mirrors) == 0 {
return nil, nil
}
mirrors := make([]Writer, 0, len(cfg.Mirrors))
for i, m := range cfg.Mirrors {
if m.Type == BackendTypeGit {
return nil, fmt.Errorf("checkpoints.mirrors[%d]: a %q mirror would duplicate the primary ref and is not supported", i, BackendTypeGit)
}
store, err := build(ctx, env, m.Type, m.Config)
if err != nil {
return nil, fmt.Errorf("checkpoints.mirrors[%d]: %w", i, err)
}
mirrors = append(mirrors, store)
}
return mirrors, nil
}
func resolveOpenRefs(ctx context.Context, opts OpenOptions) PersistentRefs {
if opts.Refs != nil {
return *opts.Refs
}
// .... remaining code ....