Delete forks that a restart interrupted

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)

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 = currentForkOwner()
 	newMeta := &metadata{StoredMetadata: forkMeta}
 	if err := m.saveForkMetadata(ctx, newMeta); err != nil {
 		return nil, false, err
@@ -561,6 +563,84 @@ 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 != nil {
+		meta.ForkPending = nil
+		if err := m.saveMetadata(meta); err != nil {
+			return fmt.Errorf("save completed fork metadata: %w", err)
+		}
+	}
+	inst.ForkPending = nil
+	return nil
+}
+
+// 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/<pid>/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()
+	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 == 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 {
+			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 +746,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 = nil
 	return dst
 }

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")
+}

lib/instances/fork_pending_test.go

@@ -0,0 +1,109 @@
+package instances
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"testing"
+	"time"
+
+	"github.com/kernel/hypeman/lib/hypervisor"
+	"github.com/stretchr/testify/assert"
+	"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()
+
+	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)
+	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.Nil(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.Nil(t, fork.ForkPending, fork.Name)
+		stored, err := mgr.loadMetadata(fork.Id)
+		require.NoError(t, err)
+		assert.Nil(t, stored.ForkPending, fork.Name)
+	}
+}
+
+func TestRecoverInterruptedForksDeletesOnlyForksWhoseProcessIsGone(t *testing.T) {
+	mgr, _ := setupTestManager(t)
+	ctx := context.Background()
+
+	save := func(id string, pending *ForkOwner) {
+		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", 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")
+}

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

lib/instances/snapshot.go

@@ -558,6 +558,7 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS
 		}
 	}
 
+	forkMeta.ForkPending = currentForkOwner()
 	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)

lib/instances/types.go

@@ -97,6 +97,13 @@ 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, 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