amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostbox
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "log/slog"
10 "os"
11 "path/filepath"
12 "sort"
13 "strings"
14
15 "github.com/benbjohnson/litestream"
16 "github.com/carlosframework/platform/internal/genselect"
17 "github.com/carlosframework/platform/internal/hostwrite"
18 "github.com/carlosframework/platform/internal/lease"
19 "github.com/superfly/ltx"
20)
21
22// RetireArtifacts moves a predecessor installation out of the live namespace.
23// It is called only when the durable local generation differs from the selected
24// generation, and while Work still vouches for the installation being replaced.
25type RetireArtifacts func(context.Context, string) error
26
27// ColdRestore installs a selected immutable generation. It never accepts an
28// unselected current object and never reuses a local database.
29type ColdRestore struct {
30 Reader genselect.Reader
31 Bucket string
32 Generation string // newly admitted installation, distinct from baseline plan.Generation
33 Work hostwrite.Work
34 Identity hostwrite.Identity
35 RetireArtifacts RetireArtifacts
36 // BeforeLegacyInstall parks an unmarked source outside Work.With. The
37 // caller must bind it for serving, including a genuinely empty creator.
38 BeforeLegacyInstall func(context.Context, string, string) error
39 // BeforeInstallLocked validates enrollment after the marker recheck. An
40 // error prevents purge, temporary database creation and marker writes.
41 BeforeInstallLocked func(context.Context, string, string) error
42}
43
44// Install purges the old local state and restores plan into dbPath. All local
45// mutation stays inside the work selected after admission.
46func (r ColdRestore) Install(ctx context.Context, dbPath string, plan *genselect.Plan) error {
47 if ctx == nil || r.Reader == nil || r.Bucket == "" || r.Work == nil || plan == nil ||
48 plan.Generation == "" || !safeGeneration(r.Generation) || len(plan.Files) == 0 || plan.MaxTXID == 0 || dbPath == "" {
49 return ErrUnsupported
50 }
51 var capturedGeneration string
52 if err := r.Work.With(ctx, r.Identity, func() error {
53 var err error
54 capturedGeneration, err = readGeneration(dbPath + ".gen")
55 return err
56 }); err != nil {
57 return fmt.Errorf("capture local generation: %w", err)
58 }
59 files, finalChecksum, err := loadPinned(ctx, r.Reader, r.Bucket, plan)
60 if err != nil {
61 return err
62 }
63 // Retirement takes park.Local's own flock. Calling it inside Work.With
64 // would recursively acquire the same lock and deadlock.
65 if capturedGeneration != "" && capturedGeneration != r.Generation {
66 if r.RetireArtifacts == nil {
67 return fmt.Errorf("retire generation %q: %w", capturedGeneration, ErrUnsupported)
68 }
69 if err := r.RetireArtifacts(ctx, capturedGeneration); err != nil {
70 return fmt.Errorf("retire generation %q: %w", capturedGeneration, err)
71 }
72 }
73 if capturedGeneration == "" && r.BeforeLegacyInstall != nil {
74 if err := r.BeforeLegacyInstall(ctx, dbPath, r.Generation); err != nil {
75 return err
76 }
77 }
78 return r.Work.With(ctx, r.Identity, func() error {
79 currentGeneration, err := readGeneration(dbPath + ".gen")
80 if err != nil {
81 return fmt.Errorf("recheck local generation: %w", err)
82 }
83 if capturedGeneration != "" && capturedGeneration != r.Generation {
84 // Successful predecessor retirement normally moves the enrolled
85 // marker with the rest of the installation. Absence is therefore the
86 // expected receipt disposition. A different marker means a successor
87 // appeared after our capture and must never be purged.
88 if currentGeneration != "" {
89 return ErrChanged
90 }
91 } else if currentGeneration != capturedGeneration {
92 return ErrChanged
93 }
94 if r.BeforeInstallLocked != nil {
95 if err := r.BeforeInstallLocked(ctx, dbPath, r.Generation); err != nil {
96 return err
97 }
98 }
99 if err := purgeDB(dbPath); err != nil {
100 return err
101 }
102 tmp, err := os.CreateTemp(filepath.Dir(dbPath), ".restore-*.db")
103 if err != nil {
104 return fmt.Errorf("create restore temp: %w", err)
105 }
106 tmpPath := tmp.Name()
107 if err := tmp.Close(); err != nil {
108 _ = os.Remove(tmpPath)
109 return err
110 }
111 _ = os.Remove(tmpPath) // Litestream requires an absent output.
112 // The check below opens the temp database read-only; a restored database
113 // is in WAL mode, so that open attaches a -wal and a -shm it cannot
114 // delete on close. Remove exactly the names derived from this restore's
115 // own temp path, never a glob: another restore's .restore-* files may be
116 // in flight beside it.
117 defer removeRestoreTemp(tmpPath)
118 client := &pinnedClient{files: files}
119 replica := litestream.NewReplicaWithClient(nil, client)
120 opt := litestream.NewRestoreOptions()
121 opt.OutputPath = tmpPath
122 opt.TXID = ltx.TXID(plan.MaxTXID)
123 if err := replica.Restore(ctx, opt); err != nil {
124 return fmt.Errorf("restore selected generation: %w", err)
125 }
126 f, err := os.Open(tmpPath)
127 if err != nil {
128 return err
129 }
130 checksum, checkErr := ltx.ChecksumReader(f, files[0].infoPageSize)
131 closeErr := f.Close()
132 if checkErr != nil {
133 return fmt.Errorf("checksum restored database: %w", checkErr)
134 }
135 if closeErr != nil {
136 return closeErr
137 }
138 // HeaderFlagNoChecksum is Litestream's pinned on-disk format. Its LTX
139 // trailer still detects corruption across the complete file, but is not
140 // an authenticity proof and intentionally has no database checksum to
141 // compare. A moves-owned object digest must supply authenticity. When
142 // the selected chain carries a database checksum, require the restored
143 // database to match it.
144 if finalChecksum != 0 && checksum != finalChecksum {
145 return fmt.Errorf("restored database checksum %s, want %s", checksum, finalChecksum)
146 }
147 if err := lease.CheckDBContext(ctx, tmpPath); err != nil {
148 return fmt.Errorf("check restored database: %w", err)
149 }
150 // The check may attach a WAL and a shared-memory file; neither is part
151 // of the restore's result and neither may follow the rename.
152 removeRestoreSidecars(tmpPath)
153 if err := syncFile(tmpPath); err != nil {
154 return err
155 }
156 if err := os.Rename(tmpPath, dbPath); err != nil {
157 return fmt.Errorf("install restored database: %w", err)
158 }
159 if err := writeGeneration(dbPath+".gen", r.Generation); err != nil {
160 return fmt.Errorf("install generation marker: %w", err)
161 }
162 return syncDir(filepath.Dir(dbPath))
163 })
164}
165
166// InstallEmpty installs a separately proved initial-empty admission. It writes
167// no database and no zero-TXID LTX. Authority must prove both the complete
168// empty remote baseline and first-creator status before local mutation.
169func (r ColdRestore) InstallEmpty(ctx context.Context, dbPath string, authority func(context.Context) error) error {
170 if ctx == nil || r.Work == nil || !safeGeneration(r.Generation) || dbPath == "" || authority == nil {
171 return ErrUnsupported
172 }
173 if err := authority(ctx); err != nil {
174 return err
175 }
176 var captured string
177 if err := r.Work.With(ctx, r.Identity, func() error { var e error; captured, e = readGeneration(dbPath + ".gen"); return e }); err != nil {
178 return err
179 }
180 if captured != "" && captured != r.Generation {
181 if r.RetireArtifacts == nil {
182 return ErrUnsupported
183 }
184 if err := r.RetireArtifacts(ctx, captured); err != nil {
185 return err
186 }
187 }
188 if captured == "" && r.BeforeLegacyInstall != nil {
189 if err := r.BeforeLegacyInstall(ctx, dbPath, r.Generation); err != nil {
190 return err
191 }
192 }
193 return r.Work.With(ctx, r.Identity, func() error {
194 current, err := readGeneration(dbPath + ".gen")
195 if err != nil {
196 return err
197 }
198 if captured != "" && captured != r.Generation {
199 if current != "" {
200 return ErrChanged
201 }
202 } else if current != captured {
203 return ErrChanged
204 }
205 if r.BeforeInstallLocked != nil {
206 if err := r.BeforeInstallLocked(ctx, dbPath, r.Generation); err != nil {
207 return err
208 }
209 }
210 if err := purgeDB(dbPath); err != nil {
211 return err
212 }
213 if err := writeGeneration(dbPath+".gen", r.Generation); err != nil {
214 return err
215 }
216 return syncDir(filepath.Dir(dbPath))
217 })
218}
219
220func safeGeneration(s string) bool {
221 return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00\r\n")
222}
223
224// removeRestoreSidecars removes the -wal and -shm files a check of the temp
225// database at tmpPath may have attached. Best effort, as in
226// moves.BuildGeneration: a name that is already absent is the normal case.
227func removeRestoreSidecars(tmpPath string) {
228 for _, p := range []string{tmpPath + "-wal", tmpPath + "-shm"} {
229 _ = os.Remove(p)
230 }
231}
232
233// removeRestoreTemp is the failure-path cleanup: the temp database and its
234// sidecars, by the names this restore derived and no others.
235func removeRestoreTemp(tmpPath string) {
236 _ = os.Remove(tmpPath)
237 removeRestoreSidecars(tmpPath)
238}
239
240func purgeDB(path string) error {
241 // ".seed" is the activator's seed marker (activator.SeedMarkerSuffix),
242 // spelled out because the activator imports this package: a marker
243 // must not outlive the database it named (re-review finding 8).
244 for _, p := range []string{path, path + "-wal", path + "-shm", path + ".gen", path + ".seed"} {
245 if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) {
246 return fmt.Errorf("purge %s: %w", p, err)
247 }
248 }
249 for _, p := range []string{
250 path + "-litestream",
251 filepath.Join(filepath.Dir(path), ".activator-"+filepath.Base(path)+"-litestream"),
252 } {
253 if err := os.RemoveAll(p); err != nil {
254 return fmt.Errorf("purge %s: %w", p, err)
255 }
256 }
257 return nil
258}
259
260// WriteGenerationMarker writes the <dbPath>.gen marker beside a database:
261// the prepared generation's name as plain text, the one format the cold
262// restore's readGeneration accepts. It is the shared writer the moves
263// landing uses for the database it installs (moves Plan 2 ruling G2; the
264// marker the moves spec §5.1 draws as JSON is written in this form instead,
265// and the incarnation and location it would have carried are vouched for by
266// the host guard record and the claim). A name readGeneration would refuse
267// is refused here, with nothing written.
268func WriteGenerationMarker(path, generation string) error {
269 if !safeGeneration(generation) {
270 return fmt.Errorf("hostbox: %q is not a generation marker: %w", generation, ErrUnsupported)
271 }
272 return writeGeneration(path, generation)
273}
274
275// ReadGenerationMarker reads a marker WriteGenerationMarker wrote. An
276// absent marker is "" with no error (the origin).
277func ReadGenerationMarker(path string) (string, error) { return readGeneration(path) }
278
279func readGeneration(path string) (string, error) {
280 b, err := os.ReadFile(path)
281 if errors.Is(err, os.ErrNotExist) {
282 return "", nil
283 }
284 if err != nil {
285 return "", err
286 }
287 g := strings.TrimSpace(string(b))
288 if g == "" || strings.ContainsAny(g, "/\\\x00\r\n") {
289 return "", errors.New("invalid generation marker")
290 }
291 return g, nil
292}
293
294func writeGeneration(path, generation string) error {
295 tmp, err := os.CreateTemp(filepath.Dir(path), ".generation-*")
296 if err != nil {
297 return err
298 }
299 name := tmp.Name()
300 defer os.Remove(name)
301 if _, err = io.WriteString(tmp, generation+"\n"); err == nil {
302 err = tmp.Sync()
303 }
304 if closeErr := tmp.Close(); err == nil {
305 err = closeErr
306 }
307 if err != nil {
308 return err
309 }
310 return os.Rename(name, path)
311}
312
313func syncFile(path string) error {
314 f, err := os.Open(path)
315 if err != nil {
316 return err
317 }
318 defer f.Close()
319 return f.Sync()
320}
321func syncDir(path string) error {
322 f, err := os.Open(path)
323 if err != nil {
324 return err
325 }
326 defer f.Close()
327 return f.Sync()
328}
329
330type pinnedFile struct {
331 plan genselect.File
332 data []byte
333 info *ltx.FileInfo
334 infoPageSize int
335}
336
337func loadPinned(ctx context.Context, reader genselect.Reader, bucket string, plan *genselect.Plan) ([]pinnedFile, ltx.Checksum, error) {
338 files := make([]pinnedFile, 0, len(plan.Files))
339 var last ltx.Checksum
340 var previous uint64
341 for i, selected := range plan.Files {
342 if i != 0 && (previous == ^uint64(0) || !ltx.IsContiguous(ltx.TXID(previous), ltx.TXID(selected.MinTXID), ltx.TXID(selected.MaxTXID))) {
343 return nil, 0, genselect.ErrGap
344 }
345 rc, err := reader.OpenVersion(ctx, bucket, selected.Key, selected.VersionID)
346 if err != nil {
347 return nil, 0, fmt.Errorf("open selected %s@%s: %w", selected.Key, selected.VersionID, err)
348 }
349 data, readErr := io.ReadAll(rc)
350 closeErr := rc.Close()
351 if readErr != nil {
352 return nil, 0, fmt.Errorf("read selected %s: %w", selected.Key, readErr)
353 }
354 if closeErr != nil {
355 return nil, 0, fmt.Errorf("close selected %s: %w", selected.Key, closeErr)
356 }
357 if int64(len(data)) != selected.Size {
358 return nil, 0, fmt.Errorf("selected %s size %d, want %d", selected.Key, len(data), selected.Size)
359 }
360 dec := ltx.NewDecoder(bytes.NewReader(data))
361 if err := dec.DecodeHeader(); err != nil {
362 return nil, 0, fmt.Errorf("decode %s header: %w", selected.Key, err)
363 }
364 h := dec.Header()
365 if uint64(h.MinTXID) != selected.MinTXID || uint64(h.MaxTXID) != selected.MaxTXID {
366 return nil, 0, fmt.Errorf("selected %s range mismatch", selected.Key)
367 }
368 // Litestream emits HeaderFlagNoChecksum but copies the last input's
369 // database checksum. Re-encode one page at a time for that format,
370 // retaining the original checksum for the final restored database.
371 // Nothing uses the converted bytes until the original proof passes.
372 var encoded bytes.Buffer
373 var enc *ltx.Encoder
374 if !h.NoChecksum() {
375 enc, err = ltx.NewEncoder(&encoded)
376 if err != nil {
377 return nil, 0, err
378 }
379 normalized := h
380 normalized.Flags |= ltx.HeaderFlagNoChecksum
381 normalized.PreApplyChecksum = 0
382 if err = enc.EncodeHeader(normalized); err != nil {
383 return nil, 0, err
384 }
385 }
386 buf := make([]byte, h.PageSize)
387 var ph ltx.PageHeader
388 for {
389 err = dec.DecodePage(&ph, buf)
390 if errors.Is(err, io.EOF) {
391 break
392 }
393 if err != nil {
394 return nil, 0, fmt.Errorf("decode %s: %w", selected.Key, err)
395 }
396 if enc != nil {
397 if err := enc.EncodePage(ph, buf); err != nil {
398 return nil, 0, err
399 }
400 }
401 }
402 if err := dec.Close(); err != nil {
403 return nil, 0, fmt.Errorf("verify %s trailer: %w", selected.Key, err)
404 }
405 t := dec.Trailer()
406 if i == 0 && plan.Pin != nil && plan.Pin.SnapshotCRC64 != 0 && uint64(t.FileChecksum) != plan.Pin.SnapshotCRC64 {
407 return nil, 0, genselect.ErrPinnedMismatch
408 }
409 if enc != nil {
410 if err := enc.Close(); err != nil {
411 return nil, 0, err
412 }
413 data = encoded.Bytes()
414 }
415 info := &ltx.FileInfo{Level: selected.Level, MinTXID: h.MinTXID, MaxTXID: h.MaxTXID, PreApplyChecksum: h.PreApplyChecksum, PostApplyChecksum: t.PostApplyChecksum, Size: int64(len(data))}
416 files = append(files, pinnedFile{plan: selected, data: data, info: info, infoPageSize: int(h.PageSize)})
417 last, previous = t.PostApplyChecksum, selected.MaxTXID
418 }
419 if previous != plan.MaxTXID {
420 return nil, 0, fmt.Errorf("selected final TXID %d, want %d", previous, plan.MaxTXID)
421 }
422 return files, last, nil
423}
424
425type pinnedClient struct{ files []pinnedFile }
426
427func (*pinnedClient) Type() string { return "selected-versions" }
428func (*pinnedClient) Init(context.Context) error { return nil }
429func (*pinnedClient) SetLogger(*slog.Logger) {}
430func (c *pinnedClient) LTXFiles(_ context.Context, level int, seek ltx.TXID, _ bool) (ltx.FileIterator, error) {
431 var a []*ltx.FileInfo
432 for _, f := range c.files {
433 if f.info.Level == level && f.info.MaxTXID >= seek {
434 cp := *f.info
435 a = append(a, &cp)
436 }
437 }
438 sort.Slice(a, func(i, j int) bool { return a[i].MinTXID < a[j].MinTXID })
439 return ltx.NewFileInfoSliceIterator(a), nil
440}
441func (c *pinnedClient) OpenLTXFile(_ context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
442 for _, f := range c.files {
443 if f.info.Level == level && f.info.MinTXID == min && f.info.MaxTXID == max {
444 if offset < 0 || size < 0 || offset > int64(len(f.data)) {
445 return nil, os.ErrInvalid
446 }
447 end := int64(len(f.data))
448 if size > 0 && offset+size < end {
449 end = offset + size
450 }
451 return io.NopCloser(bytes.NewReader(f.data[offset:end])), nil
452 }
453 }
454 return nil, os.ErrNotExist
455}
456func (*pinnedClient) WriteLTXFile(context.Context, int, ltx.TXID, ltx.TXID, io.Reader) (*ltx.FileInfo, error) {
457 return nil, ErrUnsupported
458}
459func (*pinnedClient) DeleteLTXFiles(context.Context, []*ltx.FileInfo) error { return ErrUnsupported }
460func (*pinnedClient) DeleteAll(context.Context) error { return ErrUnsupported }
461