From f69954f6bfde298800c5373bd4af463ba4958533 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Sun, 11 Oct 2026 08:17:58 +0000 Subject: [PATCH] Reduce microsandbox checkpoint archive I/O --- .../internal/execution/runtime_compute.go | 19 ++++- .../execution/runtime_compute_wake.go | 16 +++- .../internal/execution/runtime_lifecycle.go | 2 + services/core/internal/sandbox/node/agent.go | 21 +++++- .../tools/microsandbox-provider/README.md | 8 +- .../tools/microsandbox-provider/archive.go | 30 ++++---- .../tools/microsandbox-provider/bootstrap.go | 3 +- .../tools/microsandbox-provider/deployment.go | 23 ++++++ .../microsandbox-provider/deployment_test.go | 16 ++++ .../tools/microsandbox-provider/snapshot.go | 74 ++++++++++++++----- 10 files changed, 171 insertions(+), 41 deletions(-) diff --git a/services/core/internal/execution/runtime_compute.go b/services/core/internal/execution/runtime_compute.go index 71567378d..f9823e611 100644 --- a/services/core/internal/execution/runtime_compute.go +++ b/services/core/internal/execution/runtime_compute.go @@ -7,6 +7,7 @@ import ( "time" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/providercontract" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" @@ -43,8 +44,20 @@ func (r *runtimeLifecycle) saveCompute(ctx context.Context, owner deployment.All } idleTimeout = policy.IdleTimeout } - return r.deployment.SetCompute(ctx, owner, phase, raw, until, idleTimeout) + next, err := r.deployment.SetCompute(ctx, owner, phase, raw, until, idleTimeout) + if err == nil { + log.Ctx(ctx).Info("Runtime compute phase saved", "allocation_id", owner.ID, "node_id", owner.NodeID, "from", owner.ComputePhase, "to", phase) + } + return next, err +} + +// Durations use this Core process's monotonic clock and cover the provider or +// daemon round trip as seen by the lifecycle, including node queueing. +func logComputeStep(ctx context.Context, owner deployment.Allocation, step string, started time.Time, err error) { + log.Ctx(ctx).Info("Runtime compute step", "allocation_id", owner.ID, "node_id", owner.NodeID, "step", step, + "duration_ms", time.Since(started).Milliseconds(), "succeeded", err == nil) } + func (r *runtimeLifecycle) enableCompute(ctx context.Context, owner deployment.Allocation) (err error) { defer func() { err = withObservationOwner(owner, err) }() p := r.config.Provider @@ -181,7 +194,9 @@ func (r *runtimeLifecycle) idleCompute(ctx context.Context, p sandbox.SandboxPro func (r *runtimeLifecycle) captureCompute(ctx context.Context, p sandbox.SandboxProvider, owner deployment.Allocation, state runtimeCompute, observeOnly bool) (err error) { defer func() { err = withObservationOwner(owner, err) }() + started := time.Now() result, err := p.Suspend(ctx, sandbox.SuspendRequest{Reference: runtimeReference(owner), OperationID: state.SuspendID, Source: state.Current, Snapshot: state.Snapshot, ObserveOnly: observeOnly}) + logComputeStep(ctx, owner, "suspend", started, err) if err != nil { return err } @@ -281,7 +296,9 @@ func (r *runtimeLifecycle) restoreCompute(ctx context.Context, p sandbox.Sandbox return err } } + started := time.Now() result, err := p.Resume(ctx, sandbox.ResumeRequest{Workspace: workspace, Reference: runtimeReference(owner), OperationID: state.RestoreID, Snapshot: *state.Snapshot, Target: *state.Target, ObserveOnly: observeOnly}) + logComputeStep(ctx, owner, "resume", started, err) if err != nil { return err } diff --git a/services/core/internal/execution/runtime_compute_wake.go b/services/core/internal/execution/runtime_compute_wake.go index 4f836dc6a..2ed5a9961 100644 --- a/services/core/internal/execution/runtime_compute_wake.go +++ b/services/core/internal/execution/runtime_compute_wake.go @@ -16,7 +16,10 @@ import ( func (r *runtimeLifecycle) wakeCompute(ctx context.Context, p sandbox.SandboxProvider, owner deployment.Allocation, state runtimeCompute) (err error) { defer func() { err = withObservationOwner(owner, err) }() if state.Rollback { - if _, err := p.ResumeCompute(ctx, runtimeReference(owner), state.Current); err != nil { + started := time.Now() + _, err := p.ResumeCompute(ctx, runtimeReference(owner), state.Current) + logComputeStep(ctx, owner, "resume_compute", started, err) + if err != nil { return err } } @@ -27,13 +30,16 @@ func (r *runtimeLifecycle) wakeCompute(ctx context.Context, p sandbox.SandboxPro } // This idempotent control signal is fenced by guest PID/start time and the // suspension token. It cannot execute or replay an agent request. + started := time.Now() result, err := p.RunCommandCompute(ctx, runtimeReference(owner), state.Current, sandbox.Command{Args: []string{"oac-daemon", "resume", "--control-file", "/run/oac/daemon-suspend.json", "--environment-id", owner.EnvironmentID, "--suspend-id", state.SuspendID}}) + logComputeStep(ctx, owner, "wake_command", started, err) if err != nil { return err } if result.ExitCode != 0 { return sandbox.ErrComputeUnconfirmed } + started = time.Now() timer := time.NewTicker(100 * time.Millisecond) defer timer.Stop() for { @@ -47,8 +53,11 @@ func (r *runtimeLifecycle) wakeCompute(ctx context.Context, p sandbox.SandboxPro case <-timer.C: } } + logComputeStep(ctx, owner, "peer_authorized", started, nil) } + started := time.Now() result, err := peer.SuspendControl(ctx, proto.TypeEnvironmentResume, proto.EnvironmentSuspendPayload{EnvironmentID: owner.EnvironmentID, SuspendID: state.SuspendID, Rollback: state.Rollback}) + logComputeStep(ctx, owner, "environment_resume", started, err) if err != nil { return err } @@ -58,7 +67,10 @@ func (r *runtimeLifecycle) wakeCompute(ctx context.Context, p sandbox.SandboxPro // The artifact has been consumed. Never restore it after this generation // admits work, even if garbage collection or the final database commit fails. if state.Snapshot != nil { - if err := ignoreComputeAbsent(p.DeleteSnapshot(ctx, runtimeReference(owner), *state.Snapshot)); err != nil { + started = time.Now() + err := ignoreComputeAbsent(p.DeleteSnapshot(ctx, runtimeReference(owner), *state.Snapshot)) + logComputeStep(ctx, owner, "delete_snapshot", started, err) + if err != nil { return err } } diff --git a/services/core/internal/execution/runtime_lifecycle.go b/services/core/internal/execution/runtime_lifecycle.go index a4a14cc11..28b8dfb30 100644 --- a/services/core/internal/execution/runtime_lifecycle.go +++ b/services/core/internal/execution/runtime_lifecycle.go @@ -245,10 +245,12 @@ func (r *runtimeLifecycle) provision(ctx context.Context, tenant, environment, p if err := r.lease.CheckOwnership(ctx); err != nil { return owner, err } + started := time.Now() info, err := provider.Create(ctx, sandbox.Bootstrap{ Reference: runtimeReference(owner), SessionID: owner.SessionID, DeviceID: owner.DeviceID, Workspace: workspace, CoreURL: r.config.CoreURL, Credential: token, NetworkAccess: placement.NetworkAccess, AllowedDomains: placement.AllowedDomains, }) + logComputeStep(ctx, owner, "create", started, err) if info.Reference == runtimeReference(owner) && info.CreateSettled && info.State == "absent" { record, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) released, releaseErr := r.deployment.ReleaseAbsentCreation(record, owner) diff --git a/services/core/internal/sandbox/node/agent.go b/services/core/internal/sandbox/node/agent.go index 4a5c4abbf..b16a78d1d 100644 --- a/services/core/internal/sandbox/node/agent.go +++ b/services/core/internal/sandbox/node/agent.go @@ -3,6 +3,7 @@ package node import ( "context" "errors" + "log/slog" "math/rand/v2" "net/http" "net/url" @@ -71,6 +72,7 @@ type work struct { provider sandbox.SandboxProvider release func() ready bool + queued time.Time } // Run owns one persistent node identity and reconnects its transport only. It @@ -299,7 +301,7 @@ func (a *agent) connect(ctx context.Context) error { _ = current.write(frame{Type: "response", Response: &response{ID: q.ID, ConnectionID: current.id, ErrorCode: "invalid"}}) continue } - task := work{request: q, connection: current, provider: a.config.Provider, ready: a.ready.Load()} + task := work{request: q, connection: current, provider: a.config.Provider, ready: a.ready.Load(), queued: time.Now()} if a.config.Generations != nil { task.provider, task.ready, task.release, err = a.config.Generations.Acquire(q.DeploymentGeneration) if err != nil { @@ -387,8 +389,25 @@ func (a *agent) executeWork(ctx context.Context, task work) { // Detached native completion retains its local provider reference across cancellation. operationCtx, cancel := context.WithDeadline(context.Background(), task.request.deadline) a.active.Add(1) + started := time.Now() result := execute(operationCtx, task.provider, task.request) + finished := time.Now() a.active.Add(-1) cancel() _ = task.connection.write(frame{Type: "response", Response: &result}) + logOperation(ctx, task, started, finished, result.ErrorCode) +} + +// Queue wait and execution share this node's monotonic clock. Read-only +// observations repeat on every reconciliation tick and log at debug level. +func logOperation(ctx context.Context, task work, started, finished time.Time, code string) { + level := slog.LevelInfo + switch task.request.Operation { + case "info", "renew", "initial", "new_compute", "compute", "observe": + if code == "" { + level = slog.LevelDebug + } + } + log.Bg().Log(ctx, level, "sandbox node operation", "operation", task.request.Operation, "allocation_id", task.request.Reference.AllocationID, + "queue_ms", started.Sub(task.queued).Milliseconds(), "duration_ms", finished.Sub(started).Milliseconds(), "error_code", code) } diff --git a/services/core/tools/microsandbox-provider/README.md b/services/core/tools/microsandbox-provider/README.md index f30db5987..66f170841 100644 --- a/services/core/tools/microsandbox-provider/README.md +++ b/services/core/tools/microsandbox-provider/README.md @@ -12,7 +12,7 @@ Before every operation the helper checks that it was built with the exact offici The runtime home must be private (mode 0700), short, on local persistent storage and used by no other installation, profile or manual lifecycle tool. microsandbox uses Unix sockets there, so the node installer refuses a home whose path would exceed their limit. The home holds confidential VM disks, memory snapshots and SDK state; preserve it with the node identity and Core's database when recovering a host. Every managed lifecycle change goes through the provider. -The Runtime image is an immutable `repository@sha256:<64 lowercase hex>` reference that matches the saved Runtime release; bare image IDs and mutable tags are rejected. The node installer imports the distribution's image under that reference. To load an image by hand, `msb image load --tag repository@sha256:` must register the digest reference explicitly, with the manifest digest from `image inspect`, not the Docker image config ID. The image carries the daemon, Python 3, the native Harnesses and the shared Runtime helpers. The provider installs no registry credentials. +The Runtime image is an immutable `repository@sha256:<64 lowercase hex>` reference that matches the saved Runtime release; bare image IDs and mutable tags are rejected. The node installer imports the distribution's image under that reference. To load an image by hand, `msb image load --tag repository@sha256:` must register the digest reference explicitly, with the manifest digest from `image inspect`, not the Docker image config ID. The image carries the daemon, Python 3, the native Harnesses and the shared Runtime helpers. The provider installs no registry credentials. Creation uses the SDK's `never` pull policy. The checkpoint root is a separate adapter-owned directory, mode 0700, with a private `.oac-checkpoint-store` UUID marker. It is never derived from a workspace binding and must not be reachable through any guest mount. Archives contain confidential RAM and root-disk state. The marker and installation identity define the artifact domain; native release and binary hashes, Runtime image, kernel, CPU features and guest geometry define the conservative execution class. Filesystem paths, hostnames and local SDK homes do not define compatibility. Nodes can share this private root at different mount paths; the backing store must preserve durable writes, atomic rename and cross-host file locks. The installer's local default only advertises its own domain. @@ -30,13 +30,13 @@ A helper response carries `CreateSettled` with a configuration rejection only af Core persists operation IDs, source and target generations, exact identities and snapshot evidence before it depends on them. `Initial` and `NewCompute` only construct references and allocate nothing. -- **Suspend** pauses the exact VM, captures and verifies the full checkpoint closure, then exports a standalone SDK archive with its image into the private checkpoint root. The adapter synchronizes the archive and its digest, size and ownership receipt before force-stopping the exact paused source, removing its writable disks and deleting its local SDK snapshot. Only after that cleanup is synchronized does `SourceStopped` authorize transfer. A matching artifact is completed instead of captured again; a running source is never killed to finish an old capture. +- **Suspend** pauses the exact VM, captures and verifies the full checkpoint closure, then exports an SDK archive without the Runtime image into the private checkpoint root; the execution class pins that image, which every compatible node holds. The adapter hashes the archive once as it publishes it and synchronizes the archive and its digest, size and ownership receipt before force-stopping the exact paused source, removing its writable disks and deleting its local SDK snapshot. Only after that cleanup is synchronized does `SourceStopped` authorize transfer. A matching artifact is completed instead of captured again; a running source is never killed to finish an old capture. - **KillCompute** fences future admission for the exact target and checks its incarnation before stopping and removing it. Under that dispatch fence the SDK serializes against native launch, waits for the runtime lifecycle lock on Kill, and rechecks the terminal incarnation under that lock on Remove. A final typed absence permits a durable cleanup receipt, including for an admitted restore whose launcher died. This proves the writer is gone, never that it did not execute. -- **Restore** requires the source-settled archive receipt, matching artifact domain and execution class, and the complete archive hash before importing through the SDK and creating the precommitted target name. An existing target is adopted only when its immutable ID, if known, and its persisted `snapshot_parent` agree. Upstream restore defaults to public networking, so restore passes the same explicit host policy as creation, and no undeclared host resource or mount is inherited. +- **Restore** requires the source-settled archive receipt, matching artifact domain and execution class, and the complete archive hash before importing through the SDK, then the imported snapshot's full verification. Before native restore, the snapshot's captured image digest must equal the pinned Runtime image digest and that image must be in the local image store; otherwise restore fails as invalid without starting the target. This metadata check does not prove that every cached image object is present. The pinned SDK's restore has no pull-policy option and uses `if-missing`; missing cache objects can trigger a registry request pinned to the captured manifest digest. A failed resolution leaves the restore unresolved for exact observation and cleanup; it never substitutes another image or cold-boots the Session. Restore then creates the precommitted target name. An existing target is adopted only when its immutable ID, if known, and its persisted `snapshot_parent` agree. Upstream restore defaults to public networking, so restore passes the same explicit host policy as creation, and no undeclared host resource or mount is inherited. - Native restore leaves the managed root size unset because the target inherits the verified full snapshot. The helper accepts that only with a matching snapshot resource proof and the exact source and target identities, and it checks the target's CPU, memory and Environment disk before keeping the inherited proof. A missing root size never counts as unlimited capacity, and retained state is never resized. - Fresh restore and retry share one completion: verify the original artifact and resource proof, inspect the running target's resources and ancestry, persist its missing derived resource-proof label, then strictly reread the same native ID ([`restore_completion.go`](restore_completion.go)). A conflicting proof is an error. Native restore does not copy the source's ownership labels; ancestry supplies that evidence. There is no ordinary Start, replacement, disk-only restore or cold boot. - **ResumeCompute** thaws the same resident source after an aborted suspension. The pinned SDK handle method is name-based; the allocation lock and ID checks before and after the call fence every managed replacement. Manual lifecycle changes in the managed namespace are unsupported. -- **DeleteSnapshot** accepts only the derived operation selector and matching artifact identity. It fences delayed publication and restoration with a durable deletion receipt before deleting the local SDK snapshot and portable archive. An unsettled admitted restore blocks deletion. The small ownership tombstone remains; Core owns retention and cleanup order. +- **DeleteSnapshot** accepts only the derived operation selector and matching artifact identity. It compares the SDK descriptor digest, which covers the checkpoint root, instead of re-hashing payloads, so damaged content never blocks owned cleanup. It fences delayed publication and restoration with a durable deletion receipt before deleting the local SDK snapshot and portable archive. An unsettled admitted restore blocks deletion. The small ownership tombstone remains; Core owns retention and cleanup order. After a lost response Core uses `ObserveOnly`. It never starts a capture or restore. For an existing verified capture it may complete archive publication and exact paused-source cleanup, but never captures again or kills a running source. Source termination is recorded before local removal so a lost cleanup response can be settled from the archive receipt. A valid ownership receipt remains discoverable when archive bytes are missing or corrupt: observation returns its snapshot identity with unknown source state and `SourceStopped=false`, so ordinary explicit cleanup can proceed. Restore still requires complete archive verification. An interrupted restore may finish the derived proof on the exact running target, but never restarts a stopped target or restores again. diff --git a/services/core/tools/microsandbox-provider/archive.go b/services/core/tools/microsandbox-provider/archive.go index 608a5bdf7..1d1671ba9 100644 --- a/services/core/tools/microsandbox-provider/archive.go +++ b/services/core/tools/microsandbox-provider/archive.go @@ -152,51 +152,51 @@ func archiveDigest(path string) (string, int64, error) { n, e := io.Copy(h, f) return hex.EncodeToString(h.Sum(nil)), n, e } -func (b backend) publishArchive(ctx context.Context, a *sdk.SnapshotArtifact, s wire.SnapshotIdentity) (archiveReceipt, error) { +func (b backend) publishArchive(ctx context.Context, a *sdk.SnapshotArtifact, s wire.SnapshotIdentity) (archiveReceipt, bool, error) { prior, e := b.receipt(s.OperationID) if e == nil { if prior.Snapshot != s || prior.Deleted { - return prior, sandbox.ErrOwnership + return prior, false, sandbox.ErrOwnership } - return prior, nil + return prior, false, nil } if !errors.Is(e, os.ErrNotExist) { - return prior, e + return prior, false, e } dir := b.archiveDirectory(s.OperationID) if e = privateDirectory(dir); e != nil { - return prior, e + return prior, false, e } if e = syncDirectory(b.storeDirectory()); e != nil { - return prior, e + return prior, false, e } archive := filepath.Join(dir, "archive.msb") // A publication interrupted before its receipt can be rebuilt from the exact // verified native artifact. The shared allocation lock excludes consumers. if e = os.Remove(archive); e != nil && !errors.Is(e, os.ErrNotExist) { - return prior, e + return prior, false, e } - if e = a.SaveTo(ctx, archive, sdk.SnapshotSaveOptions{WithImage: true}); e != nil { - return prior, e + if e = a.SaveTo(ctx, archive, sdk.SnapshotSaveOptions{}); e != nil { + return prior, false, e } if e = os.Chmod(archive, 0600); e != nil { - return prior, e + return prior, false, e } f, e := os.Open(archive) if e != nil { - return prior, e + return prior, false, e } e = f.Sync() f.Close() if e != nil { - return prior, e + return prior, false, e } digest, size, e := archiveDigest(archive) if e != nil { - return prior, e + return prior, false, e } r := archiveReceipt{Snapshot: s, SHA256: digest, Size: size} - return r, durableJSON(filepath.Join(dir, "receipt.json"), r) + return r, true, durableJSON(filepath.Join(dir, "receipt.json"), r) } func (b backend) importArchive(ctx context.Context, want wire.SnapshotIdentity) (*sdk.SnapshotArtifact, error) { r, e := b.receipt(want.OperationID) @@ -232,7 +232,7 @@ func (b backend) importArchive(ctx context.Context, want wire.SnapshotIdentity) } func (b backend) deleteArchive(ctx context.Context, want wire.SnapshotIdentity) error { return b.deleteArchiveWithNative(want, func() error { - _, err := b.verifiedSnapshot(ctx, want) + _, err := b.ownedSnapshot(ctx, want) if err == nil { err = sdk.Snapshot.Remove(ctx, want.Reference, false) } diff --git a/services/core/tools/microsandbox-provider/bootstrap.go b/services/core/tools/microsandbox-provider/bootstrap.go index 79125a12d..928d83851 100644 --- a/services/core/tools/microsandbox-provider/bootstrap.go +++ b/services/core/tools/microsandbox-provider/bootstrap.go @@ -58,7 +58,8 @@ func (b backend) create(ctx context.Context) (wire.Response, error) { mount = sdk.Mount.Bind(b.q.Workspace.Path, sdk.MountOptions{}) } live, e := sdk.CreateSandbox(ctx, c.Name, - sdk.WithImage(b.q.Config.Image), sdk.WithMemory(b.q.Config.MemoryMiB), sdk.WithCPUs(b.q.Config.CPUs), + // Every compatible node holds the pinned image; nothing is ever pulled. + sdk.WithImage(b.q.Config.Image), sdk.WithPullPolicy(sdk.PullPolicyNever), sdk.WithMemory(b.q.Config.MemoryMiB), sdk.WithCPUs(b.q.Config.CPUs), sdk.WithMaxMemory(b.q.Config.MemoryMiB), sdk.WithMaxCPUs(b.q.Config.CPUs), sdk.WithRootDisk(sdk.RootDisk.Managed(b.q.Config.RootDiskMiB)), sdk.WithUser("1000:1000"), // The selected Environment mount keeps workspace and staging on one filesystem. diff --git a/services/core/tools/microsandbox-provider/deployment.go b/services/core/tools/microsandbox-provider/deployment.go index c768e0e59..9e2683ef1 100644 --- a/services/core/tools/microsandbox-provider/deployment.go +++ b/services/core/tools/microsandbox-provider/deployment.go @@ -3,11 +3,13 @@ package main import ( + "context" "crypto/sha256" "encoding/hex" "encoding/json" "github.com/google/uuid" "path/filepath" + "strings" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" wire "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox/microsandbox" @@ -95,6 +97,27 @@ func qualifySnapshotResources(config wire.Config, labels map[string]string) erro return nil } +// Archives exclude the image. The restored root must resolve from this node's +// installed image by the captured digest; a missing image record fails before +// Restore. This metadata check cannot verify the SDK's physical image cache. +func qualifyRuntimeImage(ctx context.Context, config wire.Config, captured string) error { + _, digest, _ := strings.Cut(config.Image, "@") + if captured != digest { + return sandbox.ErrInvalid + } + image, err := sdk.Image.Get(ctx, config.Image) + if sdk.IsKind(err, sdk.ErrImageNotFound) { + return sandbox.ErrInvalid + } + if err != nil { + return err + } + if image.ManifestDigest() != digest { + return sandbox.ErrInvalid + } + return nil +} + func workspaceLabels(workspace *wire.WorkspaceDirectory) map[string]string { if workspace == nil { return map[string]string{workspaceModeLabel: "owned"} diff --git a/services/core/tools/microsandbox-provider/deployment_test.go b/services/core/tools/microsandbox-provider/deployment_test.go index ed68e8678..efdde16d1 100644 --- a/services/core/tools/microsandbox-provider/deployment_test.go +++ b/services/core/tools/microsandbox-provider/deployment_test.go @@ -92,3 +92,19 @@ func TestSnapshotProofBindsResourcesWithoutChangingCleanupOwnership(t *testing.T t.Fatal("missing snapshot proof accepted", err) } } + +// Archives exclude the image, so Restore must find the pinned image locally. +// The real SDK's empty local registry proves absence fails typed, without a pull. +func TestRestoreRequiresPinnedLocalImage(t *testing.T) { + if !isolatedSDKTest(t) { + return + } + t.Setenv("MSB_HOME", t.TempDir()) + config, _ := deploymentFixture() + _, digest, _ := strings.Cut(config.Image, "@") + for name, captured := range map[string]string{"foreign": "sha256:" + strings.Repeat("b", 64), "unrecorded": "", "absent": digest} { + if err := qualifyRuntimeImage(t.Context(), config, captured); !errors.Is(err, sandbox.ErrInvalid) { + t.Fatalf("%s image accepted: %v", name, err) + } + } +} diff --git a/services/core/tools/microsandbox-provider/snapshot.go b/services/core/tools/microsandbox-provider/snapshot.go index 01475786b..7e89bcb86 100644 --- a/services/core/tools/microsandbox-provider/snapshot.go +++ b/services/core/tools/microsandbox-provider/snapshot.go @@ -25,8 +25,9 @@ func (b backend) snapshotLabels(operation string, source wire.Compute) map[strin return labels } -// inspectSnapshot opens only an allocation/operation-derived selector. -func (b backend) inspectSnapshot(ctx context.Context, operation string, source wire.Compute) (*sdk.SnapshotArtifact, wire.SnapshotIdentity, error) { +// openSnapshot opens only an allocation/operation-derived selector and returns +// its ownership identity without the checkpoint root, which needs Verify. +func (b backend) openSnapshot(ctx context.Context, operation string, source wire.Compute) (*sdk.SnapshotArtifact, wire.SnapshotIdentity, error) { selector := wire.SnapshotReference(b.q.Config, b.q.Reference, operation) artifact, e := sdk.Snapshot.Open(ctx, selector) if e != nil { @@ -44,21 +45,35 @@ func (b backend) inspectSnapshot(ctx context.Context, operation string, source w return nil, wire.SnapshotIdentity{}, sandbox.ErrOwnership } } + state := artifact.State() + if state.Checkpoint == nil || state.Checkpoint.CheckpointID == "" { + return nil, wire.SnapshotIdentity{}, sandbox.ErrOwnership + } + compatibility := sandbox.CheckpointCompatibility{ArtifactDomain: artifact.Labels()["io.oac.artifact_domain"], ExecutionClass: artifact.Labels()["io.oac.execution_class"]} + return artifact, wire.SnapshotIdentity{Compatibility: compatibility, Reference: selector, ID: artifact.ID(), Digest: artifact.Digest(), CheckpointID: state.Checkpoint.CheckpointID, OperationID: operation, SourceName: source.Name, SourceID: source.ID, SourceGeneration: source.Generation}, nil +} + +// inspectSnapshot verifies a fresh capture before its identity is published. +func (b backend) inspectSnapshot(ctx context.Context, operation string, source wire.Compute) (*sdk.SnapshotArtifact, wire.SnapshotIdentity, error) { + artifact, identity, e := b.openSnapshot(ctx, operation, source) + if e != nil { + return nil, wire.SnapshotIdentity{}, e + } report, e := artifact.Verify(ctx) if e != nil { return nil, wire.SnapshotIdentity{}, e } - state := artifact.State() - if report.Digest != artifact.Digest() || report.Checkpoint == nil || report.Checkpoint.Root == "" || report.Checkpoint.Kind == "" || state.Checkpoint == nil || state.Checkpoint.CheckpointID == "" { + if report.Digest != artifact.Digest() || report.Checkpoint == nil || report.Checkpoint.Root == "" || report.Checkpoint.Kind == "" { return nil, wire.SnapshotIdentity{}, sandbox.ErrOwnership } - compatibility := sandbox.CheckpointCompatibility{ArtifactDomain: artifact.Labels()["io.oac.artifact_domain"], ExecutionClass: artifact.Labels()["io.oac.execution_class"]} - identity := wire.SnapshotIdentity{Compatibility: compatibility, Reference: selector, ID: artifact.ID(), Digest: artifact.Digest(), CheckpointID: state.Checkpoint.CheckpointID, CheckpointRoot: report.Checkpoint.Root, OperationID: operation, SourceName: source.Name, SourceID: source.ID, SourceGeneration: source.Generation} + identity.CheckpointRoot = report.Checkpoint.Root if wire.ValidateSnapshot(b.q.Config, b.q.Reference, identity) != nil { return nil, wire.SnapshotIdentity{}, sandbox.ErrOwnership } return artifact, identity, nil } + +// verifiedSnapshot recomputes payload integrity before the artifact is used. func (b backend) verifiedSnapshot(ctx context.Context, want wire.SnapshotIdentity) (*sdk.SnapshotArtifact, error) { a, got, e := b.inspectSnapshot(ctx, want.OperationID, wire.Compute{Name: want.SourceName, ID: want.SourceID, Generation: want.SourceGeneration}) if e != nil { @@ -69,13 +84,31 @@ func (b backend) verifiedSnapshot(ctx context.Context, want wire.SnapshotIdentit } return a, nil } + +// ownedSnapshot matches a published identity for cleanup, where payload +// integrity is irrelevant. The SDK descriptor digest covers the checkpoint +// root, so an equal digest binds the published root without re-hashing. +func (b backend) ownedSnapshot(ctx context.Context, want wire.SnapshotIdentity) (*sdk.SnapshotArtifact, error) { + a, got, e := b.openSnapshot(ctx, want.OperationID, wire.Compute{Name: want.SourceName, ID: want.SourceID, Generation: want.SourceGeneration}) + if e != nil { + return nil, e + } + if got.Digest != want.Digest { + return nil, sandbox.ErrOwnership + } + got.CheckpointRoot = want.CheckpointRoot + if got != want || wire.ValidateSnapshot(b.q.Config, b.q.Reference, got) != nil { + return nil, sandbox.ErrOwnership + } + return a, nil +} func (b backend) suspend(ctx context.Context, q wire.SuspendRequest) (wire.State, error) { receipt, receiptErr := b.receipt(q.OperationID) if receiptErr == nil { if receipt.Deleted || receipt.Snapshot.SourceName != q.Source.Name || receipt.Snapshot.SourceID != q.Source.ID || receipt.Snapshot.SourceGeneration != q.Source.Generation || (q.Snapshot != nil && *q.Snapshot != receipt.Snapshot) { return wire.State{}, sandbox.ErrOwnership } - return b.settleSource(ctx, q, receipt) + return b.settleSource(ctx, q, receipt, false) } else if !errors.Is(receiptErr, os.ErrNotExist) { return wire.State{}, receiptErr } @@ -143,22 +176,26 @@ func (b backend) suspend(ctx context.Context, q wire.SuspendRequest) (wire.State if e = qualifySnapshotResources(b.q.Config, artifact.Labels()); e != nil { return wire.State{}, e } - receipt, e = b.publishArchive(ctx, artifact, snap) + receipt, hashed, e := b.publishArchive(ctx, artifact, snap) if e != nil { return wire.State{}, e } - return b.settleSource(ctx, q, receipt) + return b.settleSource(ctx, q, receipt, hashed) } -func (b backend) settleSource(ctx context.Context, q wire.SuspendRequest, r archiveReceipt) (wire.State, error) { - if e := b.verifyArchive(r); e != nil { - if q.ObserveOnly { - // Ownership comes from the durable receipt, independently of archive - // usability. Expose cleanup identity without authorizing source transfer, - // claiming its current state, or weakening the restore integrity check. - return wire.State{Compute: q.Source, Status: "unknown", Snapshot: &r.Snapshot}, nil +// hashed means this process has just written and hashed the archive under the +// allocation lock; an existing receipt is always checked against its bytes. +func (b backend) settleSource(ctx context.Context, q wire.SuspendRequest, r archiveReceipt, hashed bool) (wire.State, error) { + if !hashed { + if e := b.verifyArchive(r); e != nil { + if q.ObserveOnly { + // Ownership comes from the durable receipt, independently of archive + // usability. Expose cleanup identity without authorizing source transfer, + // claiming its current state, or weakening the restore integrity check. + return wire.State{Compute: q.Source, Status: "unknown", Snapshot: &r.Snapshot}, nil + } + return wire.State{}, e } - return wire.State{}, e } if r.SourceStopped { return wire.State{Compute: q.Source, Status: "suspended", BootstrapComplete: true, Snapshot: &r.Snapshot, SourceStopped: true}, nil @@ -268,6 +305,9 @@ func (b backend) resume(ctx context.Context, q wire.ResumeRequest) (wire.State, if err := qualifyWorkspace(artifact.Labels(), b.q.Workspace); err != nil { return wire.State{}, err } + if err := qualifyRuntimeImage(ctx, b.q.Config, artifact.ImageManifestDigest()); err != nil { + return wire.State{}, err + } _, _, e = b.inspectOwned(ctx, q.Target) if e == nil { return b.completeRestore(ctx, q, q.Target)