From b1374e47e95521a0ef3261f663db402ac6a530e7 Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Wed, 11 Nov 2026 08:07:48 +0000 Subject: [PATCH 1/2] Delete forks that a restart interrupted A fork's metadata is saved before the fork is brought to its target state. If Hypeman was killed in between, the fork stayed behind as an ordinary standby or paused instance: nothing marked it as unfinished, and a restore that was cut short kept its VMM, its memory and its vGPU. The caller had been disconnected without an answer and could not know the instance existed. A fork is now marked pending until its request succeeds, and pending forks are deleted at startup before the vGPU reconciler runs. --- cmd/api/main.go | 6 +++ lib/instances/fork.go | 51 +++++++++++++++++++ lib/instances/fork_pending_test.go | 81 ++++++++++++++++++++++++++++++ lib/instances/manager.go | 13 +++-- lib/instances/snapshot.go | 4 ++ lib/instances/types.go | 6 +++ 6 files changed, 156 insertions(+), 5 deletions(-) create mode 100644 lib/instances/fork_pending_test.go diff --git a/cmd/api/main.go b/cmd/api/main.go index 320a3adf7..4b4480569 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -365,6 +365,12 @@ func run() error { return fmt.Errorf("reconcile device state: %w", err) } + // A fork that a crash cut short still holds its VMM, memory and vGPU. + // Delete those before the vGPU reconciler counts their claims. + if r, ok := app.InstanceManager.(interface{ RecoverInterruptedForks(context.Context) }); ok { + r.RecoverInterruptedForks(app.Ctx) + } + logger.Info("Reconciling vGPU devices...") if r, ok := app.InstanceManager.(interface{ StartVGPUReconciler(context.Context) }); ok { r.StartVGPUReconciler(ctx) diff --git a/lib/instances/fork.go b/lib/instances/fork.go index 94e2a2132..d336e4a73 100644 --- a/lib/instances/fork.go +++ b/lib/instances/fork.go @@ -7,6 +7,7 @@ import ( "fmt" "hash/crc32" "os" + "path/filepath" "strings" "time" @@ -364,6 +365,7 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin restoreNeedsSourceLock = prepareResult.RequiresSnapshotSourceAlias } + forkMeta.ForkPending = true newMeta := &metadata{StoredMetadata: forkMeta} if err := m.saveForkMetadata(ctx, newMeta); err != nil { return nil, false, err @@ -561,6 +563,52 @@ func forkReturnNeedsGuestAgentReady(inst *Instance, guestReady bool) bool { return inst != nil && (inst.State == StateRunning || inst.State == StateInitializing) && !guestReady } +// completeFork clears the pending mark of a fork that reached its target state. +func (m *manager) completeFork(inst *Instance) error { + lock := m.getInstanceLock(inst.Id) + lock.Lock() + defer lock.Unlock() + + meta, err := m.loadMetadata(inst.Id) + if err != nil { + return fmt.Errorf("load completed fork metadata: %w", err) + } + if meta.ForkPending { + meta.ForkPending = false + if err := m.saveMetadata(meta); err != nil { + return fmt.Errorf("save completed fork metadata: %w", err) + } + } + inst.ForkPending = false + return nil +} + +// RecoverInterruptedForks deletes the forks that were still on their way to +// their target state when Hypeman last stopped. Nothing else does: such a fork +// looks like an ordinary standby or paused instance, and a restore that was +// cut short keeps its VMM, its memory and its vGPU. Call it once at startup, +// before requests are served. +func (m *manager) RecoverInterruptedForks(ctx context.Context) { + log := logger.FromContext(ctx) + files, err := m.listMetadataFiles() + if err != nil { + log.ErrorContext(ctx, "failed to list instances while recovering interrupted forks", "error", err) + return + } + for _, file := range files { + id := filepath.Base(filepath.Dir(file)) + meta, err := m.loadMetadata(id) + if err != nil || !meta.ForkPending { + continue + } + if err := m.cleanupForkInstanceOnError(ctx, id); err != nil { + log.ErrorContext(ctx, "failed to delete interrupted fork; it stays until it is deleted", "instance_id", id, "instance_name", meta.Name, "error", err) + continue + } + log.WarnContext(ctx, "deleted fork interrupted by a restart", "instance_id", id, "instance_name", meta.Name) + } +} + func (m *manager) cleanupForkInstanceOnError(ctx context.Context, forkID string) error { lock := m.getInstanceLock(forkID) lock.Lock() @@ -666,5 +714,8 @@ func cloneStoredMetadataWithoutPendingStandbyCompression(src StoredMetadata) Sto func cloneStoredMetadataForSnapshot(src StoredMetadata) StoredMetadata { dst := cloneStoredMetadataWithoutPendingStandbyCompression(src) dst.ExpiresAt = nil + // The mark belongs to the fork request, not to what is later made from + // a snapshot of its instance. + dst.ForkPending = false return dst } diff --git a/lib/instances/fork_pending_test.go b/lib/instances/fork_pending_test.go new file mode 100644 index 000000000..bad37a26f --- /dev/null +++ b/lib/instances/fork_pending_test.go @@ -0,0 +1,81 @@ +package instances + +import ( + "context" + "testing" + "time" + + "github.com/kernel/hypeman/lib/hypervisor" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestForkIsPendingOnlyUntilItsRequestSucceeds(t *testing.T) { + mgr, _ := setupTestManager(t) + ctx := context.Background() + + sourceID := "fork-pending-source" + createStoppedSnapshotSourceFixture(t, mgr, sourceID, sourceID, mgr.defaultHypervisor) + + // What a crash between saving the fork and finishing it leaves behind. + interrupted, _, err := mgr.forkInstanceFromStoppedOrStandby(ctx, sourceID, ForkInstanceRequest{Name: "fork-pending-interrupted"}, false) + require.NoError(t, err) + stored, err := mgr.loadMetadata(interrupted.Id) + require.NoError(t, err) + assert.True(t, stored.ForkPending) + + // A snapshot of it must not pass the mark on to what is restored from it. + ofInterrupted, err := mgr.CreateSnapshot(ctx, interrupted.Id, CreateSnapshotRequest{Kind: SnapshotKindStopped, Name: "fork-pending-of-interrupted"}) + require.NoError(t, err) + record, err := mgr.loadSnapshotRecord(ofInterrupted.Id) + require.NoError(t, err) + assert.False(t, record.StoredMetadata.ForkPending) + + fromInstance, err := mgr.ForkInstance(ctx, sourceID, ForkInstanceRequest{Name: "fork-pending-from-instance"}) + require.NoError(t, err) + snapshot, err := mgr.CreateSnapshot(ctx, sourceID, CreateSnapshotRequest{Kind: SnapshotKindStopped, Name: "fork-pending-snapshot"}) + require.NoError(t, err) + fromSnapshot, err := mgr.ForkSnapshot(ctx, snapshot.Id, ForkSnapshotRequest{Name: "fork-pending-from-snapshot", TargetState: StateStopped}) + require.NoError(t, err) + + for _, fork := range []*Instance{fromInstance, fromSnapshot} { + assert.False(t, fork.ForkPending, fork.Name) + stored, err := mgr.loadMetadata(fork.Id) + require.NoError(t, err) + assert.False(t, stored.ForkPending, fork.Name) + } +} + +func TestRecoverInterruptedForksDeletesOnlyPendingForks(t *testing.T) { + mgr, _ := setupTestManager(t) + ctx := context.Background() + + save := func(id string, pending bool) { + require.NoError(t, mgr.ensureDirectories(id)) + now := time.Now() + require.NoError(t, mgr.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: id, + Name: id, + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + StoppedAt: &now, + HypervisorType: hypervisor.TypeCloudHypervisor, + HypervisorVersion: "test", + SocketPath: mgr.paths.InstanceSocket(id, "cloud-hypervisor.sock"), + DataDir: mgr.paths.InstanceDir(id), + VsockCID: generateVsockCID(id), + VsockSocket: mgr.paths.InstanceVsockSocket(id), + ForkPending: pending, + }})) + } + save("fork-recover-interrupted", true) + save("fork-recover-finished", false) + + mgr.RecoverInterruptedForks(ctx) + + _, err := mgr.loadMetadata("fork-recover-interrupted") + require.ErrorIs(t, err, ErrNotFound) + assert.NoDirExists(t, mgr.paths.InstanceDir("fork-recover-interrupted")) + _, err = mgr.loadMetadata("fork-recover-finished") + require.NoError(t, err) +} diff --git a/lib/instances/manager.go b/lib/instances/manager.go index ed85f22f4..205a7fc79 100644 --- a/lib/instances/manager.go +++ b/lib/instances/manager.go @@ -556,12 +556,15 @@ func (m *manager) ForkInstance(ctx context.Context, id string, req ForkInstanceR } else { inst, err = m.applyForkTargetState(ctx, forked.Id, targetState) } - if err != nil { - if cleanupErr := m.cleanupForkInstanceOnError(ctx, forked.Id); cleanupErr != nil { - return nil, fmt.Errorf("apply fork target state: %w; additionally failed to cleanup forked instance %s: %v", err, forked.Id, cleanupErr) - } - return nil, fmt.Errorf("apply fork target state: %w", err) + } + if err == nil { + err = m.completeFork(inst) + } + if err != nil { + if cleanupErr := m.cleanupForkInstanceOnError(ctx, forked.Id); cleanupErr != nil { + return nil, fmt.Errorf("apply fork target state: %w; additionally failed to cleanup forked instance %s: %v", err, forked.Id, cleanupErr) } + return nil, fmt.Errorf("apply fork target state: %w", err) } m.notifyLifecycleEvent(ctx, LifecycleEventFork, inst) return inst, nil diff --git a/lib/instances/snapshot.go b/lib/instances/snapshot.go index dd6f6aead..6369785e0 100644 --- a/lib/instances/snapshot.go +++ b/lib/instances/snapshot.go @@ -558,6 +558,7 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS } } + forkMeta.ForkPending = true newMeta := &metadata{StoredMetadata: forkMeta} if forkMeta.MacOS != nil { if err := m.saveMacOSSnapshotFork(ctx, newMeta); err != nil { @@ -569,6 +570,9 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS cu.Release() inst, err := m.applyForkTargetState(ctx, forkID, targetState) + if err == nil { + err = m.completeFork(inst) + } if err != nil { if cleanupErr := m.cleanupForkInstanceOnError(ctx, forkID); cleanupErr != nil { return nil, fmt.Errorf("apply snapshot fork target state: %w; additionally failed to cleanup forked instance %s: %v", err, forkID, cleanupErr) diff --git a/lib/instances/types.go b/lib/instances/types.go index 9c16e6e68..27c85d0e7 100644 --- a/lib/instances/types.go +++ b/lib/instances/types.go @@ -97,6 +97,12 @@ type StoredMetadata struct { // IdentityRekeyPending marks a macOS clone whose SSH host keys still come from its // source. The next start rekeys them and clears the flag. IdentityRekeyPending bool `json:"identity_rekey_pending,omitempty"` + // ForkPending marks a fork that has not yet reached the state its caller + // asked for. It is set when the fork's metadata is first saved and cleared + // when the fork request succeeds. A fork still marked when Hypeman starts + // was cut short by a crash or kill, so its caller never got an answer; it + // is deleted then instead of being left behind holding its resources. + ForkPending bool `json:"fork_pending,omitempty"` // Resources (matching Cloud Hypervisor terminology) Size int64 // Base memory in bytes From f7b5f7dfb4dffa46fc968281ad1ad0c3b1cd6e4c Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Thu, 12 Nov 2026 07:29:05 +0000 Subject: [PATCH 2/2] Only delete a pending fork once the process making it is gone The pending mark now names the API process that is making the fork, by PID, start time and boot ID. The startup pass deletes a marked fork only when that process no longer exists. Without this, a second API process started against the same data directory, by mistake or to check a binary, would have deleted the forks that the running one was still bringing up. --- lib/instances/fork.go | 54 +++++++++++++++++++----- lib/instances/fork_pending_linux_test.go | 26 ++++++++++++ lib/instances/fork_pending_test.go | 44 +++++++++++++++---- lib/instances/snapshot.go | 2 +- lib/instances/types.go | 11 ++--- 5 files changed, 112 insertions(+), 25 deletions(-) create mode 100644 lib/instances/fork_pending_linux_test.go diff --git a/lib/instances/fork.go b/lib/instances/fork.go index d336e4a73..fcabde435 100644 --- a/lib/instances/fork.go +++ b/lib/instances/fork.go @@ -365,7 +365,7 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin restoreNeedsSourceLock = prepareResult.RequiresSnapshotSourceAlias } - forkMeta.ForkPending = true + forkMeta.ForkPending = currentForkOwner() newMeta := &metadata{StoredMetadata: forkMeta} if err := m.saveForkMetadata(ctx, newMeta); err != nil { return nil, false, err @@ -573,21 +573,49 @@ func (m *manager) completeFork(inst *Instance) error { if err != nil { return fmt.Errorf("load completed fork metadata: %w", err) } - if meta.ForkPending { - meta.ForkPending = false + if meta.ForkPending != nil { + meta.ForkPending = nil if err := m.saveMetadata(meta); err != nil { return fmt.Errorf("save completed fork metadata: %w", err) } } - inst.ForkPending = false + inst.ForkPending = nil return nil } -// RecoverInterruptedForks deletes the forks that were still on their way to -// their target state when Hypeman last stopped. Nothing else does: such a fork -// looks like an ordinary standby or paused instance, and a restore that was -// cut short keeps its VMM, its memory and its vGPU. Call it once at startup, -// before requests are served. +// ForkOwner identifies the API process that is making a fork, across PID reuse +// and host reboots. +type ForkOwner struct { + PID int `json:"pid"` + StartTime uint64 `json:"start_time,omitempty"` // from /proc//stat; 0 = unknown + BootID string `json:"boot_id,omitempty"` // Linux boot ID; empty = unknown +} + +func currentForkOwner() *ForkOwner { + pid := os.Getpid() + return &ForkOwner{PID: pid, StartTime: processStartTime(pid), BootID: hostBootID()} +} + +// gone reports whether the process that was making the fork no longer exists. +// When that cannot be told the owner counts as running, so a fork that another +// live API process is still working on is never deleted. +func (o ForkOwner) gone() bool { + if bootID := hostBootID(); o.BootID != "" && bootID != "" && o.BootID != bootID { + return true + } + if !ProcessExists(o.PID) { + return true + } + // The PID is in use. A different start time means it was recycled. + startTime := processStartTime(o.PID) + return o.StartTime != 0 && startTime != 0 && o.StartTime != startTime +} + +// RecoverInterruptedForks deletes the forks whose API process died before they +// reached their target state. Nothing else does: such a fork looks like an +// ordinary standby or paused instance, and a restore that was cut short keeps +// its VMM, its memory and its vGPU. Call it once at startup, before requests +// are served. func (m *manager) RecoverInterruptedForks(ctx context.Context) { log := logger.FromContext(ctx) files, err := m.listMetadataFiles() @@ -598,7 +626,11 @@ func (m *manager) RecoverInterruptedForks(ctx context.Context) { for _, file := range files { id := filepath.Base(filepath.Dir(file)) meta, err := m.loadMetadata(id) - if err != nil || !meta.ForkPending { + if err != nil || meta.ForkPending == nil { + continue + } + if !meta.ForkPending.gone() { + log.WarnContext(ctx, "not deleting pending fork: the process making it is still running", "instance_id", id, "instance_name", meta.Name, "owner_pid", meta.ForkPending.PID) continue } if err := m.cleanupForkInstanceOnError(ctx, id); err != nil { @@ -716,6 +748,6 @@ func cloneStoredMetadataForSnapshot(src StoredMetadata) StoredMetadata { dst.ExpiresAt = nil // The mark belongs to the fork request, not to what is later made from // a snapshot of its instance. - dst.ForkPending = false + dst.ForkPending = nil return dst } diff --git a/lib/instances/fork_pending_linux_test.go b/lib/instances/fork_pending_linux_test.go new file mode 100644 index 000000000..856d3965d --- /dev/null +++ b/lib/instances/fork_pending_linux_test.go @@ -0,0 +1,26 @@ +//go:build linux + +package instances + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A PID alone does not identify the process that was making a fork: it is +// handed out again after the process exits, and after a reboot. +func TestForkOwnerIsGoneWhenItsPIDBelongsToAnotherProcess(t *testing.T) { + current := *currentForkOwner() + require.NotZero(t, current.StartTime) + require.NotEmpty(t, current.BootID) + + recycled := current + recycled.StartTime++ + assert.True(t, recycled.gone(), "same PID, started at another time") + + beforeReboot := current + beforeReboot.BootID = "00000000-0000-0000-0000-000000000000" + assert.True(t, beforeReboot.gone(), "same PID and start time, recorded in an earlier boot") +} diff --git a/lib/instances/fork_pending_test.go b/lib/instances/fork_pending_test.go index bad37a26f..c0e6bc5f8 100644 --- a/lib/instances/fork_pending_test.go +++ b/lib/instances/fork_pending_test.go @@ -2,6 +2,8 @@ package instances import ( "context" + "os" + "os/exec" "testing" "time" @@ -10,6 +12,17 @@ import ( "github.com/stretchr/testify/require" ) +// exitedProcessOwner names a process that has already exited, the way a fork +// names the API process that crashed while making it. +func exitedProcessOwner(t *testing.T) *ForkOwner { + t.Helper() + process := exec.Command("true") + require.NoError(t, process.Start()) + owner := &ForkOwner{PID: process.Process.Pid, StartTime: processStartTime(process.Process.Pid), BootID: hostBootID()} + require.NoError(t, process.Wait()) + return owner +} + func TestForkIsPendingOnlyUntilItsRequestSucceeds(t *testing.T) { mgr, _ := setupTestManager(t) ctx := context.Background() @@ -22,14 +35,15 @@ func TestForkIsPendingOnlyUntilItsRequestSucceeds(t *testing.T) { require.NoError(t, err) stored, err := mgr.loadMetadata(interrupted.Id) require.NoError(t, err) - assert.True(t, stored.ForkPending) + require.NotNil(t, stored.ForkPending) + assert.Equal(t, os.Getpid(), stored.ForkPending.PID, "the mark must name the process making the fork") // A snapshot of it must not pass the mark on to what is restored from it. ofInterrupted, err := mgr.CreateSnapshot(ctx, interrupted.Id, CreateSnapshotRequest{Kind: SnapshotKindStopped, Name: "fork-pending-of-interrupted"}) require.NoError(t, err) record, err := mgr.loadSnapshotRecord(ofInterrupted.Id) require.NoError(t, err) - assert.False(t, record.StoredMetadata.ForkPending) + assert.Nil(t, record.StoredMetadata.ForkPending) fromInstance, err := mgr.ForkInstance(ctx, sourceID, ForkInstanceRequest{Name: "fork-pending-from-instance"}) require.NoError(t, err) @@ -39,18 +53,18 @@ func TestForkIsPendingOnlyUntilItsRequestSucceeds(t *testing.T) { require.NoError(t, err) for _, fork := range []*Instance{fromInstance, fromSnapshot} { - assert.False(t, fork.ForkPending, fork.Name) + assert.Nil(t, fork.ForkPending, fork.Name) stored, err := mgr.loadMetadata(fork.Id) require.NoError(t, err) - assert.False(t, stored.ForkPending, fork.Name) + assert.Nil(t, stored.ForkPending, fork.Name) } } -func TestRecoverInterruptedForksDeletesOnlyPendingForks(t *testing.T) { +func TestRecoverInterruptedForksDeletesOnlyForksWhoseProcessIsGone(t *testing.T) { mgr, _ := setupTestManager(t) ctx := context.Background() - save := func(id string, pending bool) { + save := func(id string, pending *ForkOwner) { require.NoError(t, mgr.ensureDirectories(id)) now := time.Now() require.NoError(t, mgr.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ @@ -68,14 +82,28 @@ func TestRecoverInterruptedForksDeletesOnlyPendingForks(t *testing.T) { ForkPending: pending, }})) } - save("fork-recover-interrupted", true) - save("fork-recover-finished", false) + save("fork-recover-interrupted", exitedProcessOwner(t)) + // Another API process on the same data directory is still making this one. + save("fork-recover-in-flight", currentForkOwner()) + save("fork-recover-finished", nil) mgr.RecoverInterruptedForks(ctx) _, err := mgr.loadMetadata("fork-recover-interrupted") require.ErrorIs(t, err, ErrNotFound) assert.NoDirExists(t, mgr.paths.InstanceDir("fork-recover-interrupted")) + + inFlight, err := mgr.loadMetadata("fork-recover-in-flight") + require.NoError(t, err, "a fork whose process is still running must be left alone") + assert.NotNil(t, inFlight.ForkPending) + _, err = mgr.loadMetadata("fork-recover-finished") require.NoError(t, err) } + +func TestForkOwnerIsGoneOnlyWhenItsProcessIs(t *testing.T) { + assert.False(t, currentForkOwner().gone(), "a running process") + assert.True(t, exitedProcessOwner(t).gone(), "an exited process") + // Without a start time or boot ID a live PID is all there is to go by. + assert.False(t, ForkOwner{PID: os.Getpid()}.gone(), "a live PID with nothing else recorded") +} diff --git a/lib/instances/snapshot.go b/lib/instances/snapshot.go index 6369785e0..d8bb9327a 100644 --- a/lib/instances/snapshot.go +++ b/lib/instances/snapshot.go @@ -558,7 +558,7 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS } } - forkMeta.ForkPending = true + forkMeta.ForkPending = currentForkOwner() newMeta := &metadata{StoredMetadata: forkMeta} if forkMeta.MacOS != nil { if err := m.saveMacOSSnapshotFork(ctx, newMeta); err != nil { diff --git a/lib/instances/types.go b/lib/instances/types.go index 27c85d0e7..9acd02d15 100644 --- a/lib/instances/types.go +++ b/lib/instances/types.go @@ -98,11 +98,12 @@ type StoredMetadata struct { // source. The next start rekeys them and clears the flag. IdentityRekeyPending bool `json:"identity_rekey_pending,omitempty"` // ForkPending marks a fork that has not yet reached the state its caller - // asked for. It is set when the fork's metadata is first saved and cleared - // when the fork request succeeds. A fork still marked when Hypeman starts - // was cut short by a crash or kill, so its caller never got an answer; it - // is deleted then instead of being left behind holding its resources. - ForkPending bool `json:"fork_pending,omitempty"` + // asked for, and names the API process that is bringing it there. It is + // set when the fork's metadata is first saved and cleared when the fork + // request succeeds. A fork still marked once that process is gone was cut + // short by a crash or kill, so its caller never got an answer; the next + // start deletes it instead of leaving it behind holding its resources. + ForkPending *ForkOwner `json:"fork_pending,omitempty"` // Resources (matching Cloud Hypervisor terminology) Size int64 // Base memory in bytes