diff --git a/cmd/core/utils.go b/cmd/core/utils.go index f85f991b..b994e808 100644 --- a/cmd/core/utils.go +++ b/cmd/core/utils.go @@ -24,12 +24,17 @@ import ( ) func FindHypervisor(ctx context.Context, conf *config.Config, ref string) (hypervisor.Hypervisor, error) { + owner, _, err := FindVM(ctx, conf, ref) + return owner, err +} + +// FindVM resolves ref to its owning hypervisor and the resolved VM. +func FindVM(ctx context.Context, conf *config.Config, ref string) (hypervisor.Hypervisor, *types.VM, error) { hypers, err := InitAllHypervisors(ctx, conf) if err != nil { - return nil, err + return nil, nil, err } - owner, _, err := resolveVMOwner(ctx, hypers, ref) - return owner, err + return resolveVMOwner(ctx, hypers, ref) } func ListAllVMs(ctx context.Context, hypers []hypervisor.Hypervisor) ([]*types.VM, error) { diff --git a/cmd/storebench/main.go b/cmd/storebench/main.go index 67ee798a..c397cf02 100644 --- a/cmd/storebench/main.go +++ b/cmd/storebench/main.go @@ -26,6 +26,20 @@ const opGet = "get" var benchPayload = json.RawMessage(`{"name":"bench","state":"running","config":{"cpu":2,"memory":1073741824}}`) +type benchConfig struct{ dir string } + +func (c benchConfig) BinaryName() string { return "bench" } +func (c benchConfig) RootDirPath() string { return c.dir } +func (c benchConfig) PIDFileName() string { return "pid" } +func (c benchConfig) TerminateGracePeriod() time.Duration { return time.Second } +func (c benchConfig) SocketWaitTimeout() time.Duration { return time.Second } +func (c benchConfig) EffectivePoolSize() int { return 4 } +func (c benchConfig) EnsureDirs() error { return nil } +func (c benchConfig) RunDir() string { return c.dir } +func (c benchConfig) LogDir() string { return c.dir } +func (c benchConfig) VMRunDir(id string) string { return filepath.Join(c.dir, id) } +func (c benchConfig) VMLogDir(id string) string { return filepath.Join(c.dir, id) } + func main() { if len(os.Args) < 3 { fmt.Fprintln(os.Stderr, "usage: storebench update|get [dir]") @@ -277,17 +291,3 @@ func argDir(i int) string { func uniquePayload(i int) json.RawMessage { return json.RawMessage(fmt.Sprintf(`{"name":"bench","state":"running","seq":%d}`, i)) } - -type benchConfig struct{ dir string } - -func (c benchConfig) BinaryName() string { return "bench" } -func (c benchConfig) RootDirPath() string { return c.dir } -func (c benchConfig) PIDFileName() string { return "pid" } -func (c benchConfig) TerminateGracePeriod() time.Duration { return time.Second } -func (c benchConfig) SocketWaitTimeout() time.Duration { return time.Second } -func (c benchConfig) EffectivePoolSize() int { return 4 } -func (c benchConfig) EnsureDirs() error { return nil } -func (c benchConfig) RunDir() string { return c.dir } -func (c benchConfig) LogDir() string { return c.dir } -func (c benchConfig) VMRunDir(id string) string { return filepath.Join(c.dir, id) } -func (c benchConfig) VMLogDir(id string) string { return filepath.Join(c.dir, id) } diff --git a/cmd/vm/commands.go b/cmd/vm/commands.go index 560f52d8..d921adf0 100644 --- a/cmd/vm/commands.go +++ b/cmd/vm/commands.go @@ -122,6 +122,14 @@ func Command(h Handler) *cobra.Command { rmCmd.Flags().Bool("force", false, "force delete running VMs (immediate SIGTERM/SIGKILL, no graceful window)") cliutil.AddOutputFlag(rmCmd) + reconcileStaleCreateCmd := &cobra.Command{ + Use: "reconcile-stale-create VM", + Short: "Reclaim an ownerless creating placeholder (refuses while a create or clone is in flight)", + Args: cobra.ExactArgs(1), + RunE: h.ReconcileStaleCreate, + } + cliutil.AddOutputFlag(reconcileStaleCreateCmd) + restoreCmd := &cobra.Command{ Use: "restore [flags] VM [SNAPSHOT]", Short: "Restore a running or stopped VM to a previous snapshot (or a directory via --from-dir)", @@ -184,6 +192,7 @@ func Command(h Handler) *cobra.Command { reseedCmd, logsCmd, rmCmd, + reconcileStaleCreateCmd, restoreCmd, hibernateCmd, debugCmd, diff --git a/cmd/vm/reconcile.go b/cmd/vm/reconcile.go new file mode 100644 index 00000000..9d6f3899 --- /dev/null +++ b/cmd/vm/reconcile.go @@ -0,0 +1,50 @@ +package vm + +import ( + "context" + "errors" + "fmt" + + "github.com/projecteru2/core/log" + "github.com/spf13/cobra" + + "github.com/cocoonstack/cocoon/cmd/cliutil" + cmdcore "github.com/cocoonstack/cocoon/cmd/core" + "github.com/cocoonstack/cocoon/hypervisor" +) + +type staleCreateResult struct { + ID string `json:"id,omitempty"` + Outcome hypervisor.StaleCreateOutcome `json:"outcome"` +} + +func (h Handler) ReconcileStaleCreate(cmd *cobra.Command, args []string) error { + ctx, conf, err := h.Init(cmd) + if err != nil { + return err + } + hyper, vm, err := cmdcore.FindVM(ctx, conf, args[0]) + if errors.Is(err, hypervisor.ErrNotFound) { + return outputStaleCreate(ctx, cmd, args[0], staleCreateResult{Outcome: hypervisor.StaleCreateNotFound}) + } + if err != nil { + return err + } + sup, ok := hyper.(hypervisor.Supervisable) + if !ok { + return fmt.Errorf("backend %s cannot reconcile stale creates", hyper.Type()) + } + outcome, err := sup.ReconcileStaleCreate(ctx, vm.ID) + if err != nil { + return fmt.Errorf("reconcile stale create: %w", err) + } + return outputStaleCreate(ctx, cmd, args[0], staleCreateResult{ID: vm.ID, Outcome: outcome}) +} + +func outputStaleCreate(ctx context.Context, cmd *cobra.Command, ref string, res staleCreateResult) error { + if done, jsonErr := cliutil.MaybeOutputJSON(cmd, res); done { + return jsonErr + } + log.WithFunc("cmd.vm.reconcileStaleCreate").Infof(ctx, "%s: %s", ref, res.Outcome) + return nil +} diff --git a/daemon/reconcile.go b/daemon/reconcile.go index e8106a98..bc054f4c 100644 --- a/daemon/reconcile.go +++ b/daemon/reconcile.go @@ -138,23 +138,17 @@ func (d *Daemon) convergeDead(ctx context.Context, b Supervisor, key watchKey, r return nil } -// collectOwnerless reclaims a creating placeholder whose owner died; a free ops lock is the proof, since create and clone hold it from prereserve through the final record commit. +// collectOwnerless routes a creating placeholder to the backend's shared stale-create reclaim; busy and raced outcomes are healthy skips. func (d *Daemon) collectOwnerless(ctx context.Context, b Supervisor, rec *hypervisor.VMRecord) error { - unlock, ok, err := d.tryLock(ctx, b, rec.ID) - if !ok { - return err - } - defer unlock() - fresh, err := b.PeekRecord(ctx, rec.ID) - if err != nil || fresh == nil || fresh.State != types.VMStateCreating { - return err - } logger := log.WithFunc("daemon.collectOwnerless") - if err := b.CollectStaleCreate(ctx, rec.ID, fresh); err != nil { + outcome, err := b.ReconcileStaleCreate(ctx, rec.ID) + if err != nil { logger.Errorf(ctx, err, "collect ownerless create %s", rec.ID) return err } - logger.Warnf(ctx, "collected ownerless creating VM %s", rec.ID) + if outcome == hypervisor.StaleCreateCollected { + logger.Warnf(ctx, "collected ownerless creating VM %s", rec.ID) + } return nil } diff --git a/daemon/reconcile_test.go b/daemon/reconcile_test.go index 178c47c7..5479cb9f 100644 --- a/daemon/reconcile_test.go +++ b/daemon/reconcile_test.go @@ -263,11 +263,6 @@ type fakeSupervisor struct { resumed []string } -func (f *fakeSupervisor) put(rec *hypervisor.VMRecord) *fakeSupervisor { - f.records[rec.ID] = rec - return f -} - func (f *fakeSupervisor) Type() string { return "fake-hv" } func (f *fakeSupervisor) ScanSupervision(context.Context) (hypervisor.SupervisionScan, error) { @@ -303,11 +298,9 @@ func (f *fakeSupervisor) ObserveVMM(_ context.Context, rec *hypervisor.VMRecord) func (f *fakeSupervisor) TryLockVMOps(_ context.Context, vmID string) (func(), bool, error) { f.mu.Lock() defer f.mu.Unlock() - if f.lockErr != nil { - return nil, false, f.lockErr - } - if _, held := f.busy[vmID]; held { - return nil, false, nil + busy, err := f.lockState(vmID) + if err != nil || busy { + return nil, false, err } return func() {}, true, nil } @@ -359,12 +352,40 @@ func (f *fakeSupervisor) RecoverTombstone(_ context.Context, vmID string) (bool, return true, nil } -func (f *fakeSupervisor) CollectStaleCreate(_ context.Context, vmID string, _ *hypervisor.VMRecord) error { +func (f *fakeSupervisor) ReconcileStaleCreate(_ context.Context, vmID string) (hypervisor.StaleCreateOutcome, error) { f.mu.Lock() defer f.mu.Unlock() + busy, err := f.lockState(vmID) + if err != nil { + return "", err + } + if busy { + return hypervisor.StaleCreateBusy, nil + } + rec := f.records[vmID] + if rec == nil { + return hypervisor.StaleCreateNotFound, nil + } + if rec.State != types.VMStateCreating { + return hypervisor.StaleCreateNotCreating, nil + } f.collected = append(f.collected, vmID) delete(f.records, vmID) - return nil + return hypervisor.StaleCreateCollected, nil +} + +// lockState reads the scripted lock condition; the caller holds f.mu. +func (f *fakeSupervisor) lockState(vmID string) (busy bool, err error) { + if f.lockErr != nil { + return false, f.lockErr + } + _, held := f.busy[vmID] + return held, nil +} + +func (f *fakeSupervisor) put(rec *hypervisor.VMRecord) *fakeSupervisor { + f.records[rec.ID] = rec + return f } func newFake() *fakeSupervisor { diff --git a/docs/cli.md b/docs/cli.md index 4258ddeb..fc00b93f 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -25,6 +25,7 @@ cocoon │ ├── reseed [--machine-id] VM Force a CRNG reseed inside the guest (fresh entropy over vsock) │ ├── logs [-f] [--tail N] VM Print the per-VM hypervisor log file │ ├── rm [flags] VM [VM...] Delete VM(s) (--force kills running VMs immediately) +│ ├── reconcile-stale-create VM Reclaim an ownerless creating placeholder (JSON outcome) │ ├── restore [flags] VM SNAP Restore a VM (running or stopped) to a snapshot │ ├── hibernate [flags] VM Atomically snapshot a running VM and stop it │ ├── status [VM...] Watch VM status in real time @@ -205,6 +206,19 @@ cocoon vm restore my-vm --from-dir /unrelated/lineage --force The dir is read-only across the call, so multiple clones of the same dir (golden image use case) are safe. Pass `--pull` if the base image's blobs may not be present locally — `EnsureImage` reads `image_blob_ids` from the envelope and pulls as needed. +### Reconcile Stale Create + +`cocoon vm reconcile-stale-create VM` reclaims a `creating` placeholder whose owning create or clone died mid-flight, freeing the record and its name. A free VM ops lock is the proof of ownerlessness — create and clone hold it from prereserve through the final record commit — so the verb never races an in-flight operation and needs no age heuristics (contrast `vm rm --force`, which queues behind a live clone and then deletes the freshly created VM). With `--output json` it prints `{"id": "...", "outcome": "..."}`: + +| Outcome | Meaning | +|------|-------------| +| `collected` | Placeholder reclaimed; record and name are free | +| `busy` | An in-flight operation owns the VM; nothing was touched — not worth retrying blindly | +| `not-creating` | The record has left the creating state; nothing was touched | +| `not-found` | No record under that ref (already collected, or never existed) | + +All four outcomes exit 0; a non-zero exit is a real failure (store I/O, an orphan VMM that would not die). At a startup reconcile, `busy` means an external create or clone legitimately owns the record: leave it un-indexed and revisit on the next pass — it either becomes a live VM or becomes collectable. The [daemon](daemon.md)'s ownerless-create reconcile and [GC](gc.md)'s `stale-creating` sweep run the same reclaim; this verb is for embedders that must clear skeletons synchronously (e.g. a startup reconcile) without waiting for either. + ### Reseed Flags Applies to `cocoon vm reseed` (forces fresh guest entropy over vsock; runs automatically best-effort after clone/restore, invoke manually to retry a failed auto-reseed): diff --git a/gc/orchestrator.go b/gc/orchestrator.go index a101756c..7d6630ed 100644 --- a/gc/orchestrator.go +++ b/gc/orchestrator.go @@ -20,11 +20,6 @@ type Orchestrator struct { // New returns an Orchestrator with no registered modules. func New() *Orchestrator { return &Orchestrator{} } -// Register is package-level because Go methods can't have type params. -func Register[S any](o *Orchestrator, m Module[S]) { - o.modules = append(o.modules, m) -} - // Run executes one GC cycle: recover tombstones by phase, then snapshot → // resolve → collect; modules revalidate every destructive decision under // their own entity locks (§5 loose-snapshot rule). @@ -71,6 +66,11 @@ func (o *Orchestrator) Run(ctx context.Context) error { return errors.Join(errs...) } +// Register is package-level because Go methods can't have type params. +func Register[S any](o *Orchestrator, m Module[S]) { + o.modules = append(o.modules, m) +} + // formatSummary renders counts as `m1=N m2=M`, sorted. func formatSummary(s map[string]int) string { if len(s) == 0 { diff --git a/hypervisor/backend.go b/hypervisor/backend.go index c77640a2..fc600da9 100644 --- a/hypervisor/backend.go +++ b/hypervisor/backend.go @@ -62,6 +62,8 @@ type BackendConfig interface { VMLogDir(id string) string } +var _ Supervisable = (*Backend)(nil) + // Backend provides shared store operations for hypervisor backends. type Backend struct { Typ string diff --git a/hypervisor/clone.go b/hypervisor/clone.go index 7416a55c..4b3300fb 100644 --- a/hypervisor/clone.go +++ b/hypervisor/clone.go @@ -43,6 +43,25 @@ func (b *Backend) CloneFromStream( }) } +// FinalizeClone persists the record and emits the clone open-interval pair. +func (b *Backend) FinalizeClone(ctx context.Context, vmID string, info *types.VM, bootCfg *types.BootConfig, blobIDs map[string]struct{}, sourceSnapshotID string) error { + if err := b.UpdateRecord(ctx, vmID, func(r *VMRecord) error { + r.VM = *info + // The subpackage building info cannot reach markTransition; without this the clone commits at generation zero. + markTransition(r, info.State, types.TransitionClone, timeNow()) + r.BootConfig = bootCfg + r.FirstBooted = true + if blobIDs != nil { + r.ImageBlobIDs = blobIDs + } + return nil + }); err != nil { + return err + } + b.emitOpenInterval(ctx, info, metering.ReasonClone, sourceSnapshotID, timeNow()) + return nil +} + func (b *Backend) cloneBase( ctx context.Context, vmID string, vmCfg *types.VMConfig, net types.NetSetup, snapshotConfig *types.SnapshotConfig, @@ -62,22 +81,3 @@ func (b *Backend) cloneBase( } return afterExtract(ctx, vmID, vmCfg, net, runDir, logDir, now, snapshotConfig.ID) } - -// FinalizeClone persists the record and emits the clone open-interval pair. -func (b *Backend) FinalizeClone(ctx context.Context, vmID string, info *types.VM, bootCfg *types.BootConfig, blobIDs map[string]struct{}, sourceSnapshotID string) error { - if err := b.UpdateRecord(ctx, vmID, func(r *VMRecord) error { - r.VM = *info - // The subpackage building info cannot reach markTransition; without this the clone commits at generation zero. - markTransition(r, info.State, types.TransitionClone, timeNow()) - r.BootConfig = bootCfg - r.FirstBooted = true - if blobIDs != nil { - r.ImageBlobIDs = blobIDs - } - return nil - }); err != nil { - return err - } - b.emitOpenInterval(ctx, info, metering.ReasonClone, sourceSnapshotID, timeNow()) - return nil -} diff --git a/hypervisor/cloudhypervisor/cloudhypervisor.go b/hypervisor/cloudhypervisor/cloudhypervisor.go index 05310d69..1cbddd57 100644 --- a/hypervisor/cloudhypervisor/cloudhypervisor.go +++ b/hypervisor/cloudhypervisor/cloudhypervisor.go @@ -5,6 +5,10 @@ import ( "fmt" "github.com/cocoonstack/cocoon/config" + "github.com/cocoonstack/cocoon/extend/disk" + "github.com/cocoonstack/cocoon/extend/fs" + "github.com/cocoonstack/cocoon/extend/netresize" + "github.com/cocoonstack/cocoon/extend/vfio" "github.com/cocoonstack/cocoon/hypervisor" "github.com/cocoonstack/cocoon/meta" "github.com/cocoonstack/cocoon/metering" @@ -15,6 +19,13 @@ const typ = "cloud-hypervisor" var ( _ hypervisor.Hypervisor = (*CloudHypervisor)(nil) _ hypervisor.Direct = (*CloudHypervisor)(nil) + _ disk.Attacher = (*CloudHypervisor)(nil) + _ disk.Lister = (*CloudHypervisor)(nil) + _ fs.Attacher = (*CloudHypervisor)(nil) + _ fs.Lister = (*CloudHypervisor)(nil) + _ vfio.Attacher = (*CloudHypervisor)(nil) + _ vfio.Lister = (*CloudHypervisor)(nil) + _ netresize.Resizer = (*CloudHypervisor)(nil) ) // CloudHypervisor implements hypervisor.Hypervisor. diff --git a/hypervisor/cloudhypervisor/extend.go b/hypervisor/cloudhypervisor/extend.go index 6b4ab8df..d0932bc2 100644 --- a/hypervisor/cloudhypervisor/extend.go +++ b/hypervisor/cloudhypervisor/extend.go @@ -13,7 +13,6 @@ import ( "github.com/cocoonstack/cocoon/extend/disk" "github.com/cocoonstack/cocoon/extend/fs" - "github.com/cocoonstack/cocoon/extend/netresize" "github.com/cocoonstack/cocoon/extend/vfio" "github.com/cocoonstack/cocoon/hypervisor" "github.com/cocoonstack/cocoon/types" @@ -22,16 +21,6 @@ import ( const chStatePaused = "Paused" -var ( - _ disk.Attacher = (*CloudHypervisor)(nil) - _ disk.Lister = (*CloudHypervisor)(nil) - _ fs.Attacher = (*CloudHypervisor)(nil) - _ fs.Lister = (*CloudHypervisor)(nil) - _ vfio.Attacher = (*CloudHypervisor)(nil) - _ vfio.Lister = (*CloudHypervisor)(nil) - _ netresize.Resizer = (*CloudHypervisor)(nil) -) - func (ch *CloudHypervisor) DiskAttach(ctx context.Context, vmRef string, spec disk.Spec) (string, error) { if err := spec.Normalize(); err != nil { return "", err diff --git a/hypervisor/gc.go b/hypervisor/gc.go index 453cbb3d..f4d9ccfb 100644 --- a/hypervisor/gc.go +++ b/hypervisor/gc.go @@ -188,7 +188,7 @@ func (b *Backend) gcCollect(ctx context.Context, ids []string, snap VMGCSnapshot if rec.State != types.VMStateCreating || !rec.UpdatedAt.Before(cutoff) { return } - if err := b.CollectStaleCreate(ctx, id, rec); err != nil { + if err := b.collectStaleCreate(ctx, id, rec); err != nil { errs = append(errs, fmt.Errorf("collect %s: %w", id, err)) return } diff --git a/hypervisor/restore.go b/hypervisor/restore.go index 55a4f85a..c28bfd2a 100644 --- a/hypervisor/restore.go +++ b/hypervisor/restore.go @@ -26,14 +26,6 @@ func (b *Backend) KillForRestore(ctx context.Context, vmID string, rec *VMRecord return nil } -// failRestore marks the VM error after a restore failure; a stopped origin (the pre-kill state) is spared so hibernate wake stays retryable — run-dir-mutating steps quarantine at their own site. -func (b *Backend) failRestore(ctx context.Context, vmID string, origin types.VMState) { - if origin == types.VMStateStopped { - return - } - b.MarkError(ctx, vmID) -} - func (b *Backend) ResolveForRestore(ctx context.Context, vmRef string) (string, *VMRecord, error) { vmID, rec, err := b.ResolveAndLoad(ctx, vmRef) if err != nil { @@ -133,6 +125,14 @@ func (b *Backend) DirectRestoreSequence(ctx context.Context, vmRef string, spec return b.restoreCore(ctx, vmID, rec, spec.VMCfg, spec.SourceSnapshotID, spec.Kill, apply, spec.AfterExtract) } +// failRestore marks the VM error after a restore failure; a stopped origin (the pre-kill state) is spared so hibernate wake stays retryable — run-dir-mutating steps quarantine at their own site. +func (b *Backend) failRestore(ctx context.Context, vmID string, origin types.VMState) { + if origin == types.VMStateStopped { + return + } + b.MarkError(ctx, vmID) +} + // restoreCore is the shared kill→emit→apply→finalize tail of both restore sequences. func (b *Backend) restoreCore( ctx context.Context, vmID string, rec *VMRecord, vmCfg *types.VMConfig, sourceSnapshotID string, diff --git a/hypervisor/state.go b/hypervisor/state.go index bfa3b5b5..3c5bbe57 100644 --- a/hypervisor/state.go +++ b/hypervisor/state.go @@ -28,36 +28,6 @@ func (b *Backend) WithRunningVM(ctx context.Context, rec *VMRecord, fn func(pid return b.withRunningVM(ctx, rec, nil, fn) } -// withRunningVM resolves liveness against the caller's /proc walk; a nil scan walks now, so batch callers turn N walks into one. -func (b *Backend) withRunningVM(ctx context.Context, rec *VMRecord, scan *utils.ProcScan, fn func(pid int) error) error { - logger := log.WithFunc(b.Typ + ".withRunningVM") - pid, pidErr := utils.ReadPIDFile(b.PIDFilePath(rec.RunDir)) - if pidErr != nil && !errors.Is(pidErr, fs.ErrNotExist) { - logger.Warnf(ctx, "read PID file: %v", pidErr) - } - sockPath := SocketPath(rec.RunDir) - if utils.VerifyProcessCmdline(pid, b.Conf.BinaryName(), sockPath) { - return fn(pid) - } - // Covers pidfile/socket cleaned up before VMM exited. Fail-closed if scan errors so callers don't treat inconclusive state as ErrNotRunning. - scanned, scanErr := b.scanFor(scan, sockPath) - if scanErr != nil { - return fmt.Errorf("vm %s: pidfile-based check failed and /proc scan errored: %w (resolve the host issue and retry)", rec.ID, scanErr) - } - if len(scanned) == 0 { - return ErrNotRunning - } - logger.Warnf(ctx, "VM %s recovered live pids %v via cmdline scan", rec.ID, scanned) - return fn(scanned[0]) -} - -func (b *Backend) scanFor(scan *utils.ProcScan, sockPath string) ([]int, error) { - if scan != nil { - return scan.Find(sockPath), nil - } - return utils.FindVMMByCmdline(b.Conf.BinaryName(), sockPath) -} - // IsAPISocketLive: (true,nil)=confirmed live; (false,nil)=ENOENT/ECONNREFUSED; (true,err)=fail-closed for unknown dial errors. func (b *Backend) IsAPISocketLive(ctx context.Context, rec *VMRecord) (bool, error) { sock := SocketPath(rec.RunDir) @@ -134,33 +104,6 @@ func (b *Backend) UpdateStates(ctx context.Context, ids []string, state types.VM }) } -// Metering entries stage inside the closure and publish only after commit, so a retried closure cannot double-emit (meta contract clause 1). -func (b *Backend) batchUpdateVMs(ctx context.Context, ids []string, mutate func(r *VMRecord, id string) []metering.Entry) error { - var emits []metering.Entry - if err := b.update(ctx, func(t *vmTx) error { - staged := []metering.Entry{} - for _, id := range ids { - r, err := t.Get(id) - if err != nil { - return err - } - if r == nil { - continue - } - staged = append(staged, mutate(r, id)...) - if err := t.Put(id, r); err != nil { - return err - } - } - emits = staged - return nil - }); err != nil { - return err - } - b.emitAll(ctx, emits) - return nil -} - // MarkError flips a single VM's state to VMStateError, logging on persist failure. func (b *Backend) MarkError(ctx context.Context, id string) { if err := b.UpdateStates(ctx, []string{id}, types.VMStateError); err != nil { @@ -168,35 +111,6 @@ func (b *Backend) MarkError(ctx context.Context, id string) { } } -// markFailedOperation records retryable network convergence after an operation may have brought plumbing up without committing a usable VMM; the daemon re-observes before acting on it. -func (b *Backend) markFailedOperation(ctx context.Context, id string, markError bool) { - ctx, cancel := detachedWrite(ctx) - defer cancel() - now := timeNow() - if err := b.update(ctx, func(t *vmTx) error { - r, err := t.Get(id) - if err != nil || r == nil { - return err - } - changed := false - if markError && r.State != types.VMStateError { - markTransition(r, types.VMStateError, types.TransitionError, now) - changed = true - } - pending := needsQuiesce(r) - if r.QuiescePending != pending { - r.QuiescePending = pending - changed = true - } - if !changed { - return nil - } - return t.Put(id, r) - }); err != nil { - log.WithFunc(b.Typ+".markFailedOperation").Errorf(ctx, err, "persist failed operation for VM %s", id) - } -} - // QuarantineVM marks the VM error and persists the quarantine reason (see VMRecord.Quarantine). func (b *Backend) QuarantineVM(ctx context.Context, id, reason string) { ctx, cancel := detachedWrite(ctx) @@ -282,6 +196,92 @@ func (b *Backend) ReconcileToRunning(ctx context.Context, id string) (uint64, er return gen, nil } +// withRunningVM resolves liveness against the caller's /proc walk; a nil scan walks now, so batch callers turn N walks into one. +func (b *Backend) withRunningVM(ctx context.Context, rec *VMRecord, scan *utils.ProcScan, fn func(pid int) error) error { + logger := log.WithFunc(b.Typ + ".withRunningVM") + pid, pidErr := utils.ReadPIDFile(b.PIDFilePath(rec.RunDir)) + if pidErr != nil && !errors.Is(pidErr, fs.ErrNotExist) { + logger.Warnf(ctx, "read PID file: %v", pidErr) + } + sockPath := SocketPath(rec.RunDir) + if utils.VerifyProcessCmdline(pid, b.Conf.BinaryName(), sockPath) { + return fn(pid) + } + // Covers pidfile/socket cleaned up before VMM exited. Fail-closed if scan errors so callers don't treat inconclusive state as ErrNotRunning. + scanned, scanErr := b.scanFor(scan, sockPath) + if scanErr != nil { + return fmt.Errorf("vm %s: pidfile-based check failed and /proc scan errored: %w (resolve the host issue and retry)", rec.ID, scanErr) + } + if len(scanned) == 0 { + return ErrNotRunning + } + logger.Warnf(ctx, "VM %s recovered live pids %v via cmdline scan", rec.ID, scanned) + return fn(scanned[0]) +} + +func (b *Backend) scanFor(scan *utils.ProcScan, sockPath string) ([]int, error) { + if scan != nil { + return scan.Find(sockPath), nil + } + return utils.FindVMMByCmdline(b.Conf.BinaryName(), sockPath) +} + +// Metering entries stage inside the closure and publish only after commit, so a retried closure cannot double-emit (meta contract clause 1). +func (b *Backend) batchUpdateVMs(ctx context.Context, ids []string, mutate func(r *VMRecord, id string) []metering.Entry) error { + var emits []metering.Entry + if err := b.update(ctx, func(t *vmTx) error { + staged := []metering.Entry{} + for _, id := range ids { + r, err := t.Get(id) + if err != nil { + return err + } + if r == nil { + continue + } + staged = append(staged, mutate(r, id)...) + if err := t.Put(id, r); err != nil { + return err + } + } + emits = staged + return nil + }); err != nil { + return err + } + b.emitAll(ctx, emits) + return nil +} + +// markFailedOperation records retryable network convergence after an operation may have brought plumbing up without committing a usable VMM; the daemon re-observes before acting on it. +func (b *Backend) markFailedOperation(ctx context.Context, id string, markError bool) { + ctx, cancel := detachedWrite(ctx) + defer cancel() + now := timeNow() + if err := b.update(ctx, func(t *vmTx) error { + r, err := t.Get(id) + if err != nil || r == nil { + return err + } + changed := false + if markError && r.State != types.VMStateError { + markTransition(r, types.VMStateError, types.TransitionError, now) + changed = true + } + pending := needsQuiesce(r) + if r.QuiescePending != pending { + r.QuiescePending = pending + changed = true + } + if !changed { + return nil + } + return t.Put(id, r) + }); err != nil { + log.WithFunc(b.Typ+".markFailedOperation").Errorf(ctx, err, "persist failed operation for VM %s", id) + } +} + // detachedWrite returns a context for bookkeeping writes that must survive caller cancellation, bounded by persistTimeout. func detachedWrite(ctx context.Context) (context.Context, context.CancelFunc) { return context.WithTimeout(context.WithoutCancel(ctx), persistTimeout) diff --git a/hypervisor/state_test.go b/hypervisor/state_test.go index 4cec69ec..e52d1904 100644 --- a/hypervisor/state_test.go +++ b/hypervisor/state_test.go @@ -620,15 +620,6 @@ func TestForcedRetryStateOps(t *testing.T) { } } -func newDiskStubConfig(t *testing.T) stubBackendConfig { - dir := t.TempDir() - return stubBackendConfig{ - rootDir: dir, - indexFile: filepath.Join(dir, "index.json"), - indexLock: filepath.Join(dir, "index.lock"), - } -} - // stubBackendConfig satisfies BackendConfig for tests that only exercise the // metering wiring; unused methods panic so accidental dependence shows up loud. type stubBackendConfig struct { @@ -668,6 +659,15 @@ func (c meteringStubConfig) RunDir() string { return c.vmRunRoot } func (c meteringStubConfig) LogDir() string { return c.vmRunRoot } +func newDiskStubConfig(t *testing.T) stubBackendConfig { + dir := t.TempDir() + return stubBackendConfig{ + rootDir: dir, + indexFile: filepath.Join(dir, "index.json"), + indexLock: filepath.Join(dir, "index.lock"), + } +} + func newMeteringTestBackend(t *testing.T) (*Backend, *meteringcapture.Recorder) { t.Helper() const typ = "test-hv" diff --git a/hypervisor/stop.go b/hypervisor/stop.go index 4381a4d4..f261d2cb 100644 --- a/hypervisor/stop.go +++ b/hypervisor/stop.go @@ -105,6 +105,15 @@ func (b *Backend) DeleteAll(ctx context.Context, refs []string, force bool, stop }) } +func (b *Backend) HandleStopResult(ctx context.Context, id, runDir string, runtimeFiles []string, shutdownErr error) error { + if shutdownErr != nil && !errors.Is(shutdownErr, ErrNotRunning) { + b.MarkError(ctx, id) + return shutdownErr + } + CleanupRuntimeFiles(ctx, runDir, runtimeFiles) + return nil +} + // deleteOneLocked is DeleteAll's per-VM body, run under the ops lock. func (b *Backend) deleteOneLocked(ctx context.Context, id string, force bool, stopLocked func(context.Context, string) error, rec *VMRecord, procScan utils.ProcScan) error { sockPath := SocketPath(rec.RunDir) @@ -156,12 +165,3 @@ func (b *Backend) deleteOneLocked(ctx context.Context, id string, force bool, st b.emitDeleteClose(ctx, id, shape, computeReason, hadRunningInterval) return nil } - -func (b *Backend) HandleStopResult(ctx context.Context, id, runDir string, runtimeFiles []string, shutdownErr error) error { - if shutdownErr != nil && !errors.Is(shutdownErr, ErrNotRunning) { - b.MarkError(ctx, id) - return shutdownErr - } - CleanupRuntimeFiles(ctx, runDir, runtimeFiles) - return nil -} diff --git a/hypervisor/supervisor.go b/hypervisor/supervisor.go index 8c962d8f..6c866027 100644 --- a/hypervisor/supervisor.go +++ b/hypervisor/supervisor.go @@ -15,7 +15,15 @@ import ( "github.com/cocoonstack/cocoon/utils" ) -var _ Supervisable = (*Backend)(nil) +const ( + StaleCreateCollected StaleCreateOutcome = "collected" // ownerless placeholder reclaimed; record and name freed + StaleCreateBusy StaleCreateOutcome = "busy" // in-flight operation owns the VM; nothing touched + StaleCreateNotCreating StaleCreateOutcome = "not-creating" // record left the creating state under the lock + StaleCreateNotFound StaleCreateOutcome = "not-found" // no record under the id +) + +// StaleCreateOutcome reports what ReconcileStaleCreate did with the record. +type StaleCreateOutcome string // Supervisable is the backend surface a resident supervisor drives. type Supervisable interface { @@ -27,7 +35,7 @@ type Supervisable interface { PeekRecord(ctx context.Context, vmID string) (*VMRecord, error) ConvergeDead(ctx context.Context, vmID string, gen uint64, observedAt time.Time) error ReconcileToRunning(ctx context.Context, vmID string) (uint64, error) - CollectStaleCreate(ctx context.Context, vmID string, rec *VMRecord) error + ReconcileStaleCreate(ctx context.Context, vmID string) (StaleCreateOutcome, error) RecoverTombstone(ctx context.Context, vmID string) (bool, error) } @@ -77,16 +85,6 @@ func (b *Backend) ObserveVMMIn(ctx context.Context, rec *VMRecord, scan utils.Pr return b.observe(ctx, rec, &scan) } -func (b *Backend) observe(ctx context.Context, rec *VMRecord, scan *utils.ProcScan) (utils.ProcRef, error) { - var ref utils.ProcRef - err := b.withRunningVM(ctx, rec, scan, func(pid int) error { - var refErr error - ref, refErr = utils.ProcRefOf(pid) - return refErr - }) - return ref, err -} - // TryLockVMOps takes the VM ops lock without blocking; ok=false with a nil error means another operation owns it. func (b *Backend) TryLockVMOps(ctx context.Context, vmID string) (unlock func(), ok bool, err error) { l, err := opsLock(b.Conf, vmID) @@ -109,33 +107,6 @@ func (b *Backend) ConvergeDead(ctx context.Context, id string, gen uint64, obser return b.QuiesceIfPending(ctx, id) } -// convergeDeadRecord commits the stop transition and publishes its staged metering entry; the quiesce it schedules is the caller's to run. -func (b *Backend) convergeDeadRecord(ctx context.Context, id string, gen uint64, observedAt time.Time) error { - var emit []metering.Entry - if err := b.update(ctx, func(t *vmTx) error { - emit = nil - r, err := t.Get(id) - if err != nil || r == nil { - return err - } - if r.TransitionGeneration != gen || !NeedsDeadConvergence(r) { - return nil - } - if hasOpenComputeInterval(r) { - emit = []metering.Entry{b.makeEntry(metering.KindVMComputeStop, id, metering.ReasonStopCrash, shapeFromConfig(r.Config), observedAt)} - } - // StoppedAt is observation time, owed even to records predating the interval bookkeeping. - r.StoppedAt = &observedAt - markTransition(r, types.VMStateStopped, types.TransitionUnexpectedExit, observedAt) - r.QuiescePending = needsQuiesce(r) - return t.Put(id, r) - }); err != nil { - return err - } - b.emitAll(ctx, emit) - return nil -} - // QuiesceIfPending runs a scheduled quiesce and clears the flag fenced on the generation it read; the caller holds the ops lock with no VMM live. func (b *Backend) QuiesceIfPending(ctx context.Context, id string) error { if b.Net == nil { @@ -161,12 +132,30 @@ func (b *Backend) QuiesceIfPending(ctx context.Context, id string) error { return b.clearQuiescePending(ctx, id, gen) } -// CollectStaleCreate reclaims an ownerless creating placeholder; the held ops lock is the proof of ownerlessness, since create and clone hold it from prereserve through the final commit. -func (b *Backend) CollectStaleCreate(ctx context.Context, id string, rec *VMRecord) error { - if err := b.ensureOrphanVMMDead(ctx, rec.RunDir); err != nil { - return fmt.Errorf("orphan vmm for %s: %w (dirs kept)", id, err) +// ReconcileStaleCreate reclaims id when it is an ownerless creating placeholder; a free ops lock is the proof of ownerlessness, since create and clone hold it from prereserve through the final record commit. +func (b *Backend) ReconcileStaleCreate(ctx context.Context, id string) (StaleCreateOutcome, error) { + unlock, ok, err := b.TryLockVMOps(ctx, id) + if err != nil { + return "", err } - return b.deleteVMProtocol(ctx, id, rec) + if !ok { + return StaleCreateBusy, nil + } + defer unlock() + rec, err := b.PeekRecord(ctx, id) + if err != nil { + return "", err + } + if rec == nil { + return StaleCreateNotFound, nil + } + if rec.State != types.VMStateCreating { + return StaleCreateNotCreating, nil + } + if err := b.collectStaleCreate(ctx, id, rec); err != nil { + return "", err + } + return StaleCreateCollected, nil } // RecoverTombstone drives an unfinished delete to completion under the held ops lock; supervision starts deletes of its own, so it must be able to finish them. @@ -174,6 +163,51 @@ func (b *Backend) RecoverTombstone(ctx context.Context, id string) (bool, error) return b.recoverVMTombstone(ctx, id) } +func (b *Backend) observe(ctx context.Context, rec *VMRecord, scan *utils.ProcScan) (utils.ProcRef, error) { + var ref utils.ProcRef + err := b.withRunningVM(ctx, rec, scan, func(pid int) error { + var refErr error + ref, refErr = utils.ProcRefOf(pid) + return refErr + }) + return ref, err +} + +// convergeDeadRecord commits the stop transition and publishes its staged metering entry; the quiesce it schedules is the caller's to run. +func (b *Backend) convergeDeadRecord(ctx context.Context, id string, gen uint64, observedAt time.Time) error { + var emit []metering.Entry + if err := b.update(ctx, func(t *vmTx) error { + emit = nil + r, err := t.Get(id) + if err != nil || r == nil { + return err + } + if r.TransitionGeneration != gen || !NeedsDeadConvergence(r) { + return nil + } + if hasOpenComputeInterval(r) { + emit = []metering.Entry{b.makeEntry(metering.KindVMComputeStop, id, metering.ReasonStopCrash, shapeFromConfig(r.Config), observedAt)} + } + // StoppedAt is observation time, owed even to records predating the interval bookkeeping. + r.StoppedAt = &observedAt + markTransition(r, types.VMStateStopped, types.TransitionUnexpectedExit, observedAt) + r.QuiescePending = needsQuiesce(r) + return t.Put(id, r) + }); err != nil { + return err + } + b.emitAll(ctx, emit) + return nil +} + +// collectStaleCreate runs the reclaim under the caller's held ops lock: no orphan VMM may survive, then the tombstoned delete protocol. +func (b *Backend) collectStaleCreate(ctx context.Context, id string, rec *VMRecord) error { + if err := b.ensureOrphanVMMDead(ctx, rec.RunDir); err != nil { + return fmt.Errorf("orphan vmm for %s: %w (dirs kept)", id, err) + } + return b.deleteVMProtocol(ctx, id, rec) +} + // clearQuiescePending is relaxed: losing the clear only costs one idempotent re-quiesce on a later pass. func (b *Backend) clearQuiescePending(ctx context.Context, id string, gen uint64) error { return b.updateRelaxed(ctx, func(t *vmTx) error { diff --git a/hypervisor/supervisor_test.go b/hypervisor/supervisor_test.go index b278810e..a39da3a5 100644 --- a/hypervisor/supervisor_test.go +++ b/hypervisor/supervisor_test.go @@ -236,7 +236,7 @@ func TestNeedsDeadConvergence(t *testing.T) { } } -func TestCollectStaleCreateFreesTheName(t *testing.T) { +func TestReconcileStaleCreateCollectsAndFreesTheName(t *testing.T) { b, _ := newMeteringTestBackend(t) ctx := t.Context() cfg := &types.VMConfig{Name: "alpha", Config: types.Config{CPU: 1}} @@ -244,8 +244,9 @@ func TestCollectStaleCreateFreesTheName(t *testing.T) { t.Fatalf("ReserveVM: %v", err) } - if err := b.CollectStaleCreate(ctx, "vm1", recordOf(t, b, "vm1")); err != nil { - t.Fatalf("CollectStaleCreate: %v", err) + outcome, err := b.ReconcileStaleCreate(ctx, "vm1") + if err != nil || outcome != StaleCreateCollected { + t.Fatalf("got (%q, %v), want collected", outcome, err) } if rec, err := b.PeekRecord(ctx, "vm1"); err != nil || rec != nil { t.Errorf("got record %v (err %v), want it collected", rec, err) @@ -255,6 +256,52 @@ func TestCollectStaleCreateFreesTheName(t *testing.T) { } } +// A held ops lock is the create owner still working; the record must survive untouched. +func TestReconcileStaleCreateRefusesInFlightCreate(t *testing.T) { + b, _ := newMeteringTestBackend(t) + ctx := t.Context() + cfg := &types.VMConfig{Name: "alpha", Config: types.Config{CPU: 1}} + if err := b.ReserveVM(ctx, "vm1", cfg, nil, t.TempDir(), t.TempDir()); err != nil { + t.Fatalf("ReserveVM: %v", err) + } + held, err := b.LockVMOps(ctx, "vm1") + if err != nil { + t.Fatalf("LockVMOps: %v", err) + } + defer held() + + outcome, err := b.ReconcileStaleCreate(ctx, "vm1") + if err != nil || outcome != StaleCreateBusy { + t.Fatalf("got (%q, %v), want busy", outcome, err) + } + if rec, err := b.PeekRecord(ctx, "vm1"); err != nil || rec == nil { + t.Errorf("got record %v (err %v), want the in-flight create untouched", rec, err) + } +} + +func TestReconcileStaleCreateRefusesNonCreating(t *testing.T) { + b, _ := newMeteringTestBackend(t) + ctx := t.Context() + seedRunningVM(t, b, "vm1", 1, 1<<30, 10<<30) + + outcome, err := b.ReconcileStaleCreate(ctx, "vm1") + if err != nil || outcome != StaleCreateNotCreating { + t.Fatalf("got (%q, %v), want not-creating", outcome, err) + } + if got := recordOf(t, b, "vm1").State; got != types.VMStateRunning { + t.Errorf("got state %q, want the record untouched at running", got) + } +} + +func TestReconcileStaleCreateReportsMissingRecord(t *testing.T) { + b, _ := newMeteringTestBackend(t) + + outcome, err := b.ReconcileStaleCreate(t.Context(), "ghost") + if err != nil || outcome != StaleCreateNotFound { + t.Fatalf("got (%q, %v), want not-found", outcome, err) + } +} + func TestTryLockVMOpsReportsBusyWithoutError(t *testing.T) { b, _ := newMeteringTestBackend(t) ctx := t.Context() diff --git a/hypervisor/teardown.go b/hypervisor/teardown.go index a70bc6db..1e523430 100644 --- a/hypervisor/teardown.go +++ b/hypervisor/teardown.go @@ -23,6 +23,21 @@ type vmCleanup struct { LogDir string `json:"log_dir,omitempty"` } +// EntryGuardLoad runs the tombstone entry guard under the caller's ops lock — +// roll a leased tombstone back, drive a deleting one to completion and refuse — +// returning the record from the guard's own transaction, sparing lock-held +// entry paths a second whole-namespace read. +func (b *Backend) EntryGuardLoad(ctx context.Context, id string) (VMRecord, error) { + rec, err := b.entryGuard(ctx, id) + if err != nil { + return VMRecord{}, err + } + if rec == nil { + return VMRecord{}, fmt.Errorf("%q not found", id) + } + return *rec, nil +} + func (b *Backend) tombstones() *tombstone.Table { return tombstone.NewTable(b.Meta, b.NS) } @@ -122,21 +137,6 @@ func (b *Backend) recoverVMTombstone(ctx context.Context, id string) (done bool, return true, nil } -// EntryGuardLoad runs the tombstone entry guard under the caller's ops lock — -// roll a leased tombstone back, drive a deleting one to completion and refuse — -// returning the record from the guard's own transaction, sparing lock-held -// entry paths a second whole-namespace read. -func (b *Backend) EntryGuardLoad(ctx context.Context, id string) (VMRecord, error) { - rec, err := b.entryGuard(ctx, id) - if err != nil { - return VMRecord{}, err - } - if rec == nil { - return VMRecord{}, fmt.Errorf("%q not found", id) - } - return *rec, nil -} - func (b *Backend) entryGuard(ctx context.Context, id string) (*VMRecord, error) { var rec *VMRecord err := b.update(ctx, func(t *vmTx) error { diff --git a/hypervisor/teardown_test.go b/hypervisor/teardown_test.go index e3b41390..7df40b28 100644 --- a/hypervisor/teardown_test.go +++ b/hypervisor/teardown_test.go @@ -287,28 +287,6 @@ func TestPrepareStartRefusesMidDeleting(t *testing.T) { } } -func seedProtoVM(t *testing.T, b *Backend, id string) *VMRecord { - t.Helper() - runDir, logDir := t.TempDir(), t.TempDir() - if err := os.WriteFile(filepath.Join(runDir, "cow.raw"), []byte("x"), 0o644); err != nil { - t.Fatal(err) - } - if err := b.ReserveVM(t.Context(), id, &types.VMConfig{Name: "proto-" + id}, nil, runDir, logDir); err != nil { - t.Fatalf("reserve: %v", err) - } - if err := b.UpdateRecord(t.Context(), id, func(r *VMRecord) error { - r.State = types.VMStateStopped - return nil - }); err != nil { - t.Fatal(err) - } - rec, err := b.LoadRecord(t.Context(), id) - if err != nil { - t.Fatal(err) - } - return &rec -} - // stubNetwork stands in for the injected host-networking seam; unset hooks are no-ops. type stubNetwork struct { recover func(context.Context, *types.VM) error @@ -337,6 +315,28 @@ func (s stubNetwork) Cleanup(ctx context.Context, vmID string) error { return s.cleanup(ctx, vmID) } +func seedProtoVM(t *testing.T, b *Backend, id string) *VMRecord { + t.Helper() + runDir, logDir := t.TempDir(), t.TempDir() + if err := os.WriteFile(filepath.Join(runDir, "cow.raw"), []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + if err := b.ReserveVM(t.Context(), id, &types.VMConfig{Name: "proto-" + id}, nil, runDir, logDir); err != nil { + t.Fatalf("reserve: %v", err) + } + if err := b.UpdateRecord(t.Context(), id, func(r *VMRecord) error { + r.State = types.VMStateStopped + return nil + }); err != nil { + t.Fatal(err) + } + rec, err := b.LoadRecord(t.Context(), id) + if err != nil { + t.Fatal(err) + } + return &rec +} + // entryGuardOnly exercises the entry guard the way an entrypoint does, discarding the record. func entryGuardOnly(ctx context.Context, b *Backend, id string) error { _, err := b.entryGuard(ctx, id) diff --git a/images/oci/image.go b/images/oci/image.go index a02863df..b16ce5fb 100644 --- a/images/oci/image.go +++ b/images/oci/image.go @@ -10,6 +10,10 @@ import ( type imageIndex = images.Index[imageEntry] +type layerEntry struct { + Digest images.Digest `json:"digest"` +} + // Paths derive from digests at runtime; not stored. type imageEntry struct { Ref string `json:"ref"` @@ -33,10 +37,6 @@ func (e imageEntry) DigestHexes() []string { return hexes } -type layerEntry struct { - Digest images.Digest `json:"digest"` -} - // normalizeRef expands a short OCI ref (e.g. "ubuntu:24.04" → "docker.io/library/ubuntu:24.04"). func normalizeRef(s string) (string, bool) { parsed, err := name.ParseReference(s) diff --git a/meta/contracttest/suite.go b/meta/contracttest/suite.go index 678266dd..c6fa4068 100644 --- a/meta/contracttest/suite.go +++ b/meta/contracttest/suite.go @@ -25,6 +25,28 @@ var errForcedRollback = errors.New("forced rollback") // Factory returns a fresh Store serving the given namespaces; the suite owns its lifetime. type Factory func(t *testing.T, namespaces []string) meta.Store +type record struct { + Name string `json:"name"` + N int `json:"n"` +} + +type retryStore struct { + meta.Store +} + +func (r *retryStore) Update(ctx context.Context, sc meta.Scope, mode meta.CommitMode, fn func(meta.Writer) error) error { + err := r.Store.Update(ctx, sc, mode, func(w meta.Writer) error { + if err := fn(w); err != nil { + return err + } + return errForcedRollback + }) + if err != nil && !errors.Is(err, errForcedRollback) { + return err + } + return r.Store.Update(ctx, sc, mode, fn) +} + // Run executes the full contract suite against factory. func Run(t *testing.T, factory Factory) { t.Run("CRUD", func(t *testing.T) { testCRUD(t, factory) }) @@ -42,11 +64,6 @@ func Run(t *testing.T, factory Factory) { // ForcedRetry wraps s so every Update closure runs twice — once rolled back, once for real — enforcing pure retryable closures. func ForcedRetry(s meta.Store) meta.Store { return &retryStore{Store: s} } -type record struct { - Name string `json:"name"` - N int `json:"n"` -} - func testCRUD(t *testing.T, factory Factory) { ctx := t.Context() s := factory(t, []string{nsAlpha}) @@ -425,23 +442,6 @@ func testLogCursor(t *testing.T, factory Factory) { } } -type retryStore struct { - meta.Store -} - -func (r *retryStore) Update(ctx context.Context, sc meta.Scope, mode meta.CommitMode, fn func(meta.Writer) error) error { - err := r.Store.Update(ctx, sc, mode, func(w meta.Writer) error { - if err := fn(w); err != nil { - return err - } - return errForcedRollback - }) - if err != nil && !errors.Is(err, errForcedRollback) { - return err - } - return r.Store.Update(ctx, sc, mode, fn) -} - func get1(t *testing.T, s meta.Store, c *meta.Collection[record]) (*record, error) { t.Helper() var rec *record diff --git a/meta/json/store.go b/meta/json/store.go index 3e87e929..51aeab3e 100644 --- a/meta/json/store.go +++ b/meta/json/store.go @@ -36,6 +36,18 @@ type Namespace struct { Codec Codec } +type nsState struct { + def Namespace + locker *flock.Lock +} + +type loaded struct { + model *Model + // recovered means main was undecodable and .prev was served; commit must + // not rotate, or it would destroy the only good generation. + recovered bool +} + var _ meta.Store = (*Store)(nil) // Store is the json engine: one flocked file per namespace, legacy write @@ -182,18 +194,67 @@ func (s *Store) loadAll(ctx context.Context, states []*nsState) (map[string]*loa return models, nil } -type nsState struct { - def Namespace - locker *flock.Lock +var _ meta.Reader = (*txReader)(nil) + +type txReader struct { + models map[string]*loaded } -type loaded struct { +func (r *txReader) GetRaw(_ context.Context, ns, table, id string) (json.RawMessage, bool, error) { + l, ok := r.models[ns] + if !ok { + return nil, false, fmt.Errorf("read %s: %w", ns, meta.ErrScope) + } + raw, ok := l.model.Get(table, id) + // Detached values (contract clause 4): aliasing model bytes would let a + // caller mutate committed state without PutRaw's scope/durability checks. + return slices.Clone(raw), ok, nil +} + +func (r *txReader) ScanRaw(_ context.Context, ns, table string, fn func(id string, raw json.RawMessage) error) error { + l, ok := r.models[ns] + if !ok { + return fmt.Errorf("read %s: %w", ns, meta.ErrScope) + } + return l.model.Scan(table, func(id string, raw json.RawMessage) error { + return fn(id, slices.Clone(raw)) + }) +} + +var _ meta.Writer = (*txWriter)(nil) + +type txWriter struct { + txReader + write string model *Model - // recovered means main was undecodable and .prev was served; commit must - // not rotate, or it would destroy the only good generation. - recovered bool + mode meta.CommitMode +} + +func (w *txWriter) PutRaw(_ context.Context, ns, table, id string, raw json.RawMessage, relaxedOK bool) error { + if err := meta.CheckWriteScope(ns, w.write, w.mode, relaxedOK); err != nil { + return err + } + w.model.Put(table, id, slices.Clone(raw)) + return nil +} + +func (w *txWriter) DeleteRaw(_ context.Context, ns, table, id string, relaxedOK bool) error { + if err := meta.CheckWriteScope(ns, w.write, w.mode, relaxedOK); err != nil { + return err + } + w.model.Delete(table, id) + return nil +} + +// coded pairs an engine error with its taxonomy sentinel without changing the message text. +type coded struct { + err error + mark error } +func (c *coded) Error() string { return c.err.Error() } +func (c *coded) Unwrap() []error { return []error{c.err, c.mark} } + // loadNamespace ports the legacy load: a missing file is empty, an // undecodable main falls back to the .prev generation, read errors fail closed. func loadNamespace(ctx context.Context, def Namespace) (*loaded, error) { @@ -296,67 +357,6 @@ func crash(step string) error { return testCrashStep(step) } -var _ meta.Reader = (*txReader)(nil) - -type txReader struct { - models map[string]*loaded -} - -func (r *txReader) GetRaw(_ context.Context, ns, table, id string) (json.RawMessage, bool, error) { - l, ok := r.models[ns] - if !ok { - return nil, false, fmt.Errorf("read %s: %w", ns, meta.ErrScope) - } - raw, ok := l.model.Get(table, id) - // Detached values (contract clause 4): aliasing model bytes would let a - // caller mutate committed state without PutRaw's scope/durability checks. - return slices.Clone(raw), ok, nil -} - -func (r *txReader) ScanRaw(_ context.Context, ns, table string, fn func(id string, raw json.RawMessage) error) error { - l, ok := r.models[ns] - if !ok { - return fmt.Errorf("read %s: %w", ns, meta.ErrScope) - } - return l.model.Scan(table, func(id string, raw json.RawMessage) error { - return fn(id, slices.Clone(raw)) - }) -} - -var _ meta.Writer = (*txWriter)(nil) - -type txWriter struct { - txReader - write string - model *Model - mode meta.CommitMode -} - -func (w *txWriter) PutRaw(_ context.Context, ns, table, id string, raw json.RawMessage, relaxedOK bool) error { - if err := meta.CheckWriteScope(ns, w.write, w.mode, relaxedOK); err != nil { - return err - } - w.model.Put(table, id, slices.Clone(raw)) - return nil -} - -func (w *txWriter) DeleteRaw(_ context.Context, ns, table, id string, relaxedOK bool) error { - if err := meta.CheckWriteScope(ns, w.write, w.mode, relaxedOK); err != nil { - return err - } - w.model.Delete(table, id) - return nil -} - -// coded pairs an engine error with its taxonomy sentinel without changing the message text. -type coded struct { - err error - mark error -} - -func (c *coded) Error() string { return c.err.Error() } -func (c *coded) Unwrap() []error { return []error{c.err, c.mark} } - func code(err, mark error) error { if errors.Is(err, syscall.ENOSPC) { mark = meta.ErrNoSpace diff --git a/meta/json/tables.go b/meta/json/tables.go index 86d326b3..fb0390db 100644 --- a/meta/json/tables.go +++ b/meta/json/tables.go @@ -23,6 +23,22 @@ type TableSpec struct { StringList bool } +var _ Codec = TableCodec{} + +// TableCodec is the declaration-only codec for pure table-shaped namespaces: +// a subsystem states its legacy field layout and owns no codec code. +type TableCodec struct { + Specs []TableSpec +} + +func (c TableCodec) Decode(data []byte) (*Model, error) { + return DecodeTables(data, c.Specs) +} + +func (c TableCodec) Encode(m *Model) ([]byte, error) { + return EncodeTables(m, c.Specs) +} + // DecodeTables loads specs' map fields into a fresh Model (sorted insertion, // matching what encoding/json always wrote). Single streaming pass — a whole-file // unmarshal into raw messages tokenizes the payload twice. @@ -115,22 +131,6 @@ func EncodeTables(m *Model, specs []TableSpec) ([]byte, error) { return append(buf, '}', '\n'), nil } -var _ Codec = TableCodec{} - -// TableCodec is the declaration-only codec for pure table-shaped namespaces: -// a subsystem states its legacy field layout and owns no codec code. -type TableCodec struct { - Specs []TableSpec -} - -func (c TableCodec) Decode(data []byte) (*Model, error) { - return DecodeTables(data, c.Specs) -} - -func (c TableCodec) Encode(m *Model) ([]byte, error) { - return EncodeTables(m, c.Specs) -} - func appendKey(buf []byte, key string) []byte { if len(buf) > 1 { buf = append(buf, ',') diff --git a/meta/sqlite/store.go b/meta/sqlite/store.go index d1448509..ff581027 100644 --- a/meta/sqlite/store.go +++ b/meta/sqlite/store.go @@ -299,6 +299,106 @@ func (s *Store) verifyIdentity() error { return nil } +var ( + _ meta.Reader = (*txHandle)(nil) + _ meta.Writer = (*txHandle)(nil) +) + +// txHandle implements Reader/Writer over one transaction; values are +// detached by construction (every read allocates from row scans). +type txHandle struct { + ctx context.Context + tx *sql.Tx + sm map[string]tableStmts + scope map[string]struct{} + write string + mode meta.CommitMode +} + +func (h *txHandle) GetRaw(ctx context.Context, ns, table, id string) (json.RawMessage, bool, error) { + if err := h.checkRead(ns); err != nil { + return nil, false, err + } + ts, err := h.stmts(ns, table) + if err != nil { + return nil, false, err + } + var data []byte + err = h.tx.StmtContext(ctx, ts.get).QueryRowContext(ctx, id).Scan(&data) + if errors.Is(err, sql.ErrNoRows) { + return nil, false, nil + } + if err != nil { + return nil, false, mapErr(err) + } + return data, true, nil +} + +func (h *txHandle) ScanRaw(ctx context.Context, ns, table string, fn func(id string, raw json.RawMessage) error) error { + if err := h.checkRead(ns); err != nil { + return err + } + ts, err := h.stmts(ns, table) + if err != nil { + return err + } + rows, err := h.tx.StmtContext(ctx, ts.scan).QueryContext(ctx) + if err != nil { + return mapErr(err) + } + defer rows.Close() //nolint:errcheck + for rows.Next() { + var id string + var data []byte + if err := rows.Scan(&id, &data); err != nil { + return mapErr(err) + } + if err := fn(id, data); err != nil { + return err + } + } + return mapErr(rows.Err()) +} + +func (h *txHandle) PutRaw(ctx context.Context, ns, table, id string, raw json.RawMessage, relaxedOK bool) error { + if err := meta.CheckWriteScope(ns, h.write, h.mode, relaxedOK); err != nil { + return err + } + ts, err := h.stmts(ns, table) + if err != nil { + return err + } + _, err = h.tx.StmtContext(ctx, ts.put).ExecContext(ctx, id, []byte(raw)) + return mapErr(err) +} + +func (h *txHandle) DeleteRaw(ctx context.Context, ns, table, id string, relaxedOK bool) error { + if err := meta.CheckWriteScope(ns, h.write, h.mode, relaxedOK); err != nil { + return err + } + ts, err := h.stmts(ns, table) + if err != nil { + return err + } + _, err = h.tx.StmtContext(ctx, ts.del).ExecContext(ctx, id) + return mapErr(err) +} + +func (h *txHandle) stmts(ns, table string) (tableStmts, error) { + ts, ok := h.sm[ns+"\x00"+table] + if !ok { + return tableStmts{}, fmt.Errorf("table %s/%s not declared: %w", ns, table, meta.ErrScope) + } + return ts, nil +} + +func (h *txHandle) checkRead(ns string) error { + if _, ok := h.scope[ns]; !ok { + return fmt.Errorf("read %s: %w", ns, meta.ErrScope) + } + return nil +} + func tableName(ns, table string) string { return quoteIdent(ns + "__" + table) } @@ -418,103 +518,3 @@ func mapErr(err error) error { return err } } - -var ( - _ meta.Reader = (*txHandle)(nil) - _ meta.Writer = (*txHandle)(nil) -) - -// txHandle implements Reader/Writer over one transaction; values are -// detached by construction (every read allocates from row scans). -type txHandle struct { - ctx context.Context - tx *sql.Tx - sm map[string]tableStmts - scope map[string]struct{} - write string - mode meta.CommitMode -} - -func (h *txHandle) GetRaw(ctx context.Context, ns, table, id string) (json.RawMessage, bool, error) { - if err := h.checkRead(ns); err != nil { - return nil, false, err - } - ts, err := h.stmts(ns, table) - if err != nil { - return nil, false, err - } - var data []byte - err = h.tx.StmtContext(ctx, ts.get).QueryRowContext(ctx, id).Scan(&data) - if errors.Is(err, sql.ErrNoRows) { - return nil, false, nil - } - if err != nil { - return nil, false, mapErr(err) - } - return data, true, nil -} - -func (h *txHandle) ScanRaw(ctx context.Context, ns, table string, fn func(id string, raw json.RawMessage) error) error { - if err := h.checkRead(ns); err != nil { - return err - } - ts, err := h.stmts(ns, table) - if err != nil { - return err - } - rows, err := h.tx.StmtContext(ctx, ts.scan).QueryContext(ctx) - if err != nil { - return mapErr(err) - } - defer rows.Close() //nolint:errcheck - for rows.Next() { - var id string - var data []byte - if err := rows.Scan(&id, &data); err != nil { - return mapErr(err) - } - if err := fn(id, data); err != nil { - return err - } - } - return mapErr(rows.Err()) -} - -func (h *txHandle) PutRaw(ctx context.Context, ns, table, id string, raw json.RawMessage, relaxedOK bool) error { - if err := meta.CheckWriteScope(ns, h.write, h.mode, relaxedOK); err != nil { - return err - } - ts, err := h.stmts(ns, table) - if err != nil { - return err - } - _, err = h.tx.StmtContext(ctx, ts.put).ExecContext(ctx, id, []byte(raw)) - return mapErr(err) -} - -func (h *txHandle) DeleteRaw(ctx context.Context, ns, table, id string, relaxedOK bool) error { - if err := meta.CheckWriteScope(ns, h.write, h.mode, relaxedOK); err != nil { - return err - } - ts, err := h.stmts(ns, table) - if err != nil { - return err - } - _, err = h.tx.StmtContext(ctx, ts.del).ExecContext(ctx, id) - return mapErr(err) -} - -func (h *txHandle) stmts(ns, table string) (tableStmts, error) { - ts, ok := h.sm[ns+"\x00"+table] - if !ok { - return tableStmts{}, fmt.Errorf("table %s/%s not declared: %w", ns, table, meta.ErrScope) - } - return ts, nil -} - -func (h *txHandle) checkRead(ns string) error { - if _, ok := h.scope[ns]; !ok { - return fmt.Errorf("read %s: %w", ns, meta.ErrScope) - } - return nil -} diff --git a/metadata/fat12.go b/metadata/fat12.go index 5695247d..06837a15 100644 --- a/metadata/fat12.go +++ b/metadata/fat12.go @@ -31,18 +31,6 @@ const ( writeBufSize = 64 << 10 ) -// CreateFAT12 streams a 1 MiB FAT12 image with VFAT long-filename support to w. -func CreateFAT12(w io.Writer, label string, files map[string][]byte) error { - b := newFAT12Builder(label) - - for _, name := range slices.Sorted(maps.Keys(files)) { - if err := b.addFile(name, files[name]); err != nil { - return err - } - } - return b.writeTo(w) -} - // fat12Builder constructs a FAT12 image in memory (FAT + root dir only) and streams the full image on writeTo. type fat12Builder struct { label string @@ -206,6 +194,18 @@ func (b *fat12Builder) makeBootSector() []byte { return boot } +// CreateFAT12 streams a 1 MiB FAT12 image with VFAT long-filename support to w. +func CreateFAT12(w io.Writer, label string, files map[string][]byte) error { + b := newFAT12Builder(label) + + for _, name := range slices.Sorted(maps.Keys(files)) { + if err := b.addFile(name, files[name]); err != nil { + return err + } + } + return b.writeTo(w) +} + // setFATEntry writes a 12-bit value into the FAT at the given cluster index. func setFATEntry(fat []byte, cluster int, val uint16) { off := cluster + cluster/2 //nolint:mnd diff --git a/progress/progress.go b/progress/progress.go index 805e45af..dd716fa9 100644 --- a/progress/progress.go +++ b/progress/progress.go @@ -8,6 +8,10 @@ type Tracker interface { OnEvent(any) } +type funcTracker func(any) + +func (f funcTracker) OnEvent(e any) { f(e) } + // NewTracker wraps a typed callback as a non-generic Tracker so Images can hold it in its interface. func NewTracker[E any](fn func(E)) Tracker { return funcTracker(func(v any) { @@ -16,7 +20,3 @@ func NewTracker[E any](fn func(E)) Tracker { } }) } - -type funcTracker func(any) - -func (f funcTracker) OnEvent(e any) { f(e) } diff --git a/snapshot/localfile/gc.go b/snapshot/localfile/gc.go index b7e35f79..1920e738 100644 --- a/snapshot/localfile/gc.go +++ b/snapshot/localfile/gc.go @@ -58,6 +58,39 @@ type snapshotGCSnapshot struct { func (s snapshotGCSnapshot) UsedBlobIDs() map[string]struct{} { return s.blobIDs } +// PinnedBlobIDs reports every image blob pinned by a snapshot record — image GC's under-digest-lock recheck source. +func (lf *LocalFile) PinnedBlobIDs(ctx context.Context) (map[string]struct{}, error) { + pins := map[string]struct{}{} + if err := lf.view(ctx, func(t *snapTx) error { + return t.Scan(func(_ string, rec *snapshot.SnapshotRecord) error { + maps.Copy(pins, rec.ImageBlobIDs) + return nil + }) + }); err != nil { + return nil, err + } + return pins, nil +} + +// gcRecover resumes existing snapshot tombstones by phase before discovery. +func (lf *LocalFile) gcRecover(ctx context.Context) []error { + var ids []string + if err := lf.view(ctx, func(t *snapTx) error { + var err error + ids, err = lf.tombstones().PendingIDs(ctx, t.Reader()) + return err + }); err != nil { + return []error{err} + } + var errs []error + for _, id := range ids { + if err := lf.recoverSnapTombstone(ctx, id); err != nil { + errs = append(errs, err) + } + } + return errs +} + func gcModule(lf *LocalFile, policy EvictionPolicy) gc.Module[snapshotGCSnapshot] { conf, recorder := lf.conf, lf.metering return gc.Module[snapshotGCSnapshot]{ @@ -194,20 +227,6 @@ func gcModule(lf *LocalFile, policy EvictionPolicy) gc.Module[snapshotGCSnapshot } } -// PinnedBlobIDs reports every image blob pinned by a snapshot record — image GC's under-digest-lock recheck source. -func (lf *LocalFile) PinnedBlobIDs(ctx context.Context) (map[string]struct{}, error) { - pins := map[string]struct{}{} - if err := lf.view(ctx, func(t *snapTx) error { - return t.Scan(func(_ string, rec *snapshot.SnapshotRecord) error { - maps.Copy(pins, rec.ImageBlobIDs) - return nil - }) - }); err != nil { - return nil, err - } - return pins, nil -} - // pickLRU maps each evict ID to its reason ("+" joins multi-match; no criteria → "lru-all"). func pickLRU(records map[string]snapshotMeta, p EvictionPolicy) map[string]string { sorted := slices.SortedFunc(maps.Keys(records), func(a, b string) int { @@ -322,22 +341,3 @@ func backfillSizeBytes(ctx context.Context, lf *LocalFile, records map[string]sn logger.Warnf(ctx, "persist backfilled SizeBytes: %v", err) } } - -// gcRecover resumes existing snapshot tombstones by phase before discovery. -func (lf *LocalFile) gcRecover(ctx context.Context) []error { - var ids []string - if err := lf.view(ctx, func(t *snapTx) error { - var err error - ids, err = lf.tombstones().PendingIDs(ctx, t.Reader()) - return err - }); err != nil { - return []error{err} - } - var errs []error - for _, id := range ids { - if err := lf.recoverSnapTombstone(ctx, id); err != nil { - errs = append(errs, err) - } - } - return errs -}