@@ -290,7 +290,19 @@ func (t *storageTemplate) Fetch(ctx context.Context, buildStore *build.DiffStore
 }
 
 func (t *storageTemplate) Close(ctx context.Context) error {
-	return closeTemplate(ctx, t)
+	err := closeTemplate(ctx, t)
+
+	// closeTemplate only removes the files it holds handles for, which leaves the
+	// metafile and the directory itself behind; nothing else reclaims them, since
+	// the startup sweep covers DefaultCacheDir and these live under
+	// TemplateCacheDir. The directory is private to this instance — CachePaths
+	// mints a fresh identifier per template — so removing it cannot touch another
+	// instance's files.
+	if pathsErr := t.paths.Close(); pathsErr != nil {
+		err = errors.Join(err, fmt.Errorf("failed to remove template cache dir: %w", pathsErr))
+	}
+
+	return err
 }
 
 func (t *storageTemplate) Files() storage.CachePaths {
@@ -340,6 +352,72 @@ func (t *storageTemplate) SchedulingMetadata(ctx context.Context) *orchestrator.
 	return scheduling.FromHeaders(rh.Metadata.BuildId, mh, rh, 0)
 }
 
+// headerFootprint reports the mapping entry count and approximate heap bytes
+// held by this template's resolved headers. It never blocks: the headers live
+// on the memfile/rootfs devices (the header holders stay unset for templates
+// loaded from storage — see SchedulingMetadata), and a device whose SetOnce has
+// not resolved yet is skipped. The gauge therefore undercounts still-fetching
+// templates rather than stalling the metrics callback on them.
+func (t *storageTemplate) headerFootprint() (entries int, bytes int) {
+	// Deduplicated by the mapping's backing storage rather than by header
+	// identity, because one allocation reaches this function by two routes. The
+	// holders below usually resolve to the very headers the devices carry; and
+	// once a snapshot's upload publishes, the device carries a CloneForUpload
+	// while the holder still carries the source — two distinct *Header sharing
+	// one Mapping, since the clone copies the struct and the copy shares its
+	// slices. Counting that allocation twice would overstate the number this
+	// gauge exists to make trustworthy, and it would do so as uploads land,
+	// which reads like retention growth rather than a counting artifact.
+	seen := make(map[*header.Header]struct{}, 4)
+	counted := make([]header.Mapping, 0, 4)
+
+	add := func(h *header.Header) {
+		if h == nil {
+			return
+		}
+		if _, dup := seen[h]; dup {
+			return
+		}
+		seen[h] = struct{}{}
+
+		for _, m := range counted {
+			if m.SharesStorageWith(h.Mapping) {
+				return
+			}
+		}
+		counted = append(counted, h.Mapping)
+
+		entries += h.Mapping.Len()
+		bytes += h.Mapping.ByteSize()
+	}
+
+	if dev, err := t.memfile.Result(); err == nil && dev != nil {
+		add(dev.Header())
+	}
+
+	if dev, err := t.rootfs.Result(); err == nil && dev != nil {
+		add(dev.Header())
+	}
+
+	// The holders are not always the headers the devices ended up on, and the
+	// difference is retained memory. A template built from a provisional memfile
+	// header keeps that header alive in memfileHeader after SwapHeaderIfCurrent
+	// has moved the device on to the deduped one, so a paused-and-deduped
+	// template holds two distinct mappings while the device reports one. Count
+	// every distinct header the template still references.
+	for _, holder := range []*utils.SetOnce[*header.Header]{t.memfileHeader, t.rootfsHeader, t.durableMemfileHeader} {
+		if holder == nil {
+			continue
+		}
+
+		if h, err := holder.Result(); err == nil {
+			add(h)
+		}
+	}
+
+	return entries, bytes
+}
+
 func (t *storageTemplate) Memfile(ctx context.Context) (block.ReadonlyDevice, error) {
 	_, span := tracer.Start(ctx, "storage-template-memfile")
 	defer span.End()
 
@@ -4,6 +4,7 @@ package template
 
 import (
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
@@ -80,3 +81,162 @@ func TestStorageTemplate_SchedulingMetadataSkipsPendingMemfile(t *testing.T) {
 	assert.Empty(t, md.GetMemfileBaseBuildId())
 	assert.Equal(t, rootfsBase.String(), md.GetRootfsBaseBuildId())
 }
+
+// footprintHeader builds a header whose mapping has exactly n entries, using a
+// distinct build id per entry so nothing merges.
+func footprintHeader(t *testing.T, n int) *header.Header {
+	t.Helper()
+
+	const blockSize = uint64(4096)
+
+	maps := make([]header.BuildMap, n)
+	for i := range maps {
+		maps[i] = header.BuildMap{
+			Offset:  uint64(i) * blockSize,
+			Length:  blockSize,
+			BuildId: uuid.New(),
+		}
+	}
+
+	h, err := header.NewHeader(&header.Metadata{
+		Version:     3,
+		BlockSize:   blockSize,
+		Size:        uint64(n) * blockSize,
+		BuildId:     uuid.New(),
+		BaseBuildId: uuid.New(),
+	}, maps)
+	require.NoError(t, err)
+	require.Equal(t, n, h.Mapping.Len())
+
+	return h
+}
+
+// The gauge exists to size a bound, so it has to see every mapping the template
+// keeps alive. After a pause the provisional header stays referenced by
+// memfileHeader while the device has already moved on to the deduped one, so a
+// gauge that reads only the devices reports one mapping where two are resident
+// — understating exactly the footprint it is used to size.
+func TestStorageTemplate_HeaderFootprintCountsRetainedHolders(t *testing.T) {
+	t.Parallel()
+
+	provisional := footprintHeader(t, 12)
+	deduped := footprintHeader(t, 3)
+	rootfsHdr := footprintHeader(t, 5)
+
+	memDev := blockmocks.NewMockReadonlyDevice(t)
+	memDev.EXPECT().Header().Return(deduped)
+	rootfsDev := blockmocks.NewMockReadonlyDevice(t)
+	rootfsDev.EXPECT().Header().Return(rootfsHdr)
+
+	tmpl := &storageTemplate{
+		memfile:              utils.NewSetOnce[block.ReadonlyDevice](),
+		rootfs:               utils.NewSetOnce[block.ReadonlyDevice](),
+		memfileHeader:        resolvedHeader(provisional),
+		rootfsHeader:         resolvedHeader(rootfsHdr),
+		durableMemfileHeader: resolvedHeader(deduped),
+	}
+	require.NoError(t, tmpl.memfile.SetValue(memDev))
+	require.NoError(t, tmpl.rootfs.SetValue(rootfsDev))
+
+	entries, bytes := tmpl.headerFootprint()
+
+	wantEntries := provisional.Mapping.Len() + deduped.Mapping.Len() + rootfsHdr.Mapping.Len()
+	wantBytes := provisional.Mapping.ByteSize() + deduped.Mapping.ByteSize() + rootfsHdr.Mapping.ByteSize()
+
+	assert.Equal(t, wantEntries, entries, "the provisional header is still resident and must be counted")
+	assert.Equal(t, wantBytes, bytes)
+}
+
+// The holders usually resolve to the same headers the devices carry. Counting
+// those twice would overstate the number the sizing decisions are made from.
+func TestStorageTemplate_HeaderFootprintCountsEachMappingOnce(t *testing.T) {
+	t.Parallel()
+
+	memHdr := footprintHeader(t, 7)
+	rootfsHdr := footprintHeader(t, 4)
+
+	memDev := blockmocks.NewMockReadonlyDevice(t)
+	memDev.EXPECT().Header().Return(memHdr)
+	rootfsDev := blockmocks.NewMockReadonlyDevice(t)
+	rootfsDev.EXPECT().Header().Return(rootfsHdr)
+
+	tmpl := &storageTemplate{
+		memfile:       utils.NewSetOnce[block.ReadonlyDevice](),
+		rootfs:        utils.NewSetOnce[block.ReadonlyDevice](),
+		memfileHeader: resolvedHeader(memHdr),
+		rootfsHeader:  resolvedHeader(rootfsHdr),
+	}
+	require.NoError(t, tmpl.memfile.SetValue(memDev))
+	require.NoError(t, tmpl.rootfs.SetValue(rootfsDev))
+
+	entries, bytes := tmpl.headerFootprint()
+
+	assert.Equal(t, memHdr.Mapping.Len()+rootfsHdr.Mapping.Len(), entries)
+	assert.Equal(t, memHdr.Mapping.ByteSize()+rootfsHdr.Mapping.ByteSize(), bytes)
+}
+
+// A template that is still fetching must contribute nothing rather than block
+// the metrics collection goroutine on its unresolved futures.
+func TestStorageTemplate_HeaderFootprintSkipsUnresolved(t *testing.T) {
+	t.Parallel()
+
+	tmpl := &storageTemplate{
+		memfile:       utils.NewSetOnce[block.ReadonlyDevice](),
+		rootfs:        utils.NewSetOnce[block.ReadonlyDevice](),
+		memfileHeader: utils.NewSetOnce[*header.Header](),
+		rootfsHeader:  utils.NewSetOnce[*header.Header](),
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+
+		entries, bytes := tmpl.headerFootprint()
+		assert.Zero(t, entries)
+		assert.Zero(t, bytes)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("headerFootprint blocked on an unresolved template")
+	}
+}
+
+// One allocation reaches headerFootprint by two routes once a snapshot's upload
+// publishes: CloneForUpload copies the Header struct, so the clone shares the
+// source's Mapping slices, and publish installs the clone on the device while
+// the holder keeps the source. Header identity does not see that, so the gauge
+// would step up as uploads land — reading like retention growth rather than a
+// counting artifact, on exactly the population a byte budget gets sized from.
+func TestStorageTemplate_HeaderFootprintCountsSharedMappingOnce(t *testing.T) {
+	t.Parallel()
+
+	source := footprintHeader(t, 9)
+	published := source.CloneForUpload(source.Metadata.Version + 1)
+	require.NotSame(t, source, published, "the clone is a distinct header")
+	require.True(t, source.Mapping.SharesStorageWith(published.Mapping),
+		"the clone must share the source's mapping, or this test proves nothing")
+
+	rootfsHdr := footprintHeader(t, 4)
+
+	memDev := blockmocks.NewMockReadonlyDevice(t)
+	memDev.EXPECT().Header().Return(published)
+	rootfsDev := blockmocks.NewMockReadonlyDevice(t)
+	rootfsDev.EXPECT().Header().Return(rootfsHdr)
+
+	tmpl := &storageTemplate{
+		memfile:       utils.NewSetOnce[block.ReadonlyDevice](),
+		rootfs:        utils.NewSetOnce[block.ReadonlyDevice](),
+		memfileHeader: resolvedHeader(source),
+		rootfsHeader:  resolvedHeader(rootfsHdr),
+	}
+	require.NoError(t, tmpl.memfile.SetValue(memDev))
+	require.NoError(t, tmpl.rootfs.SetValue(rootfsDev))
+
+	entries, bytes := tmpl.headerFootprint()
+
+	assert.Equal(t, source.Mapping.Len()+rootfsHdr.Mapping.Len(), entries,
+		"the shared mapping must be counted once, not once per header")
+	assert.Equal(t, source.Mapping.ByteSize()+rootfsHdr.Mapping.ByteSize(), bytes)
+}
@@ -108,7 +108,7 @@ type harvestInstance interface {
 
 // harvestTemplates is the subset of the template cache the harvest uses.
 type harvestTemplates interface {
-	GetTemplate(ctx context.Context, buildID string, isSnapshot, isBuilding bool, opts ...sbxtemplate.GetTemplateOpts) (sbxtemplate.Template, error)
+	GetTemplatePinned(ctx context.Context, buildID string, isSnapshot, isBuilding bool, opts ...sbxtemplate.GetTemplateOpts) (sbxtemplate.Template, func(), error)
 	UpdateMetadata(buildID string, meta metadata.Template) error
 }
 
@@ -394,12 +394,18 @@ func (h *prefetchHarvester) resumeMapping(
 	// pays no cold GCS/NFS fetch, only a local re-fault. isSnapshot=true mirrors
 	// Checkpoint's resume. The pause artifact carries no Prefetch (SameVersion
 	// dropped it), so no prefetcher runs and the trace is clean demand faults.
-	tmpl, err := h.templates.GetTemplate(ctx, buildID, true, false,
+	// Pinned: the throwaway below is a real Firecracker VM running off this
+	// template's snapfile, so for its lifetime the template must not be
+	// evictable — eviction Closes it, and Close removes the snapfile. Released
+	// last, after the reap deferred below has torn the throwaway down.
+	tmpl, releaseTemplate, err := h.templates.GetTemplatePinned(ctx, buildID, true, false,
 		sbxtemplate.GetTemplateOpts{MaxSandboxLengthHours: sbx.Config.MaxSandboxLengthHours})
 	if err != nil {
 		return nil, harvestResumeFailed, fmt.Errorf("get template: %w", err)
 	}
 
+	defer releaseTemplate()
+
 	// Throwaway identity: distinct SandboxID/ExecutionID from the (being-stopped)
 	// original so it never collides in the sandbox map. ResumeSandbox registers
 	// it in the factory's sandbox table (for network assignment and health), but
 
@@ -83,10 +83,21 @@ type fakeHarvestTemplates struct {
 	updateErr   error
 	updated     bool
 	updatedMeta metadata.Template
+
+	// pins and releases must balance: a pin the harvest never releases keeps the
+	// template resident for the life of the process.
+	pins     int
+	releases int
 }
 
-func (f *fakeHarvestTemplates) GetTemplate(context.Context, string, bool, bool, ...sbxtemplate.GetTemplateOpts) (sbxtemplate.Template, error) {
-	return nil, f.getErr
+func (f *fakeHarvestTemplates) GetTemplatePinned(context.Context, string, bool, bool, ...sbxtemplate.GetTemplateOpts) (sbxtemplate.Template, func(), error) {
+	if f.getErr != nil {
+		return nil, func() {}, f.getErr
+	}
+
+	f.pins++
+
+	return nil, func() { f.releases++ }, nil
 }
 
 func (f *fakeHarvestTemplates) UpdateMetadata(_ string, meta metadata.Template) error {
@@ -204,6 +215,8 @@ func TestHarvestRun_ConsumePersistsAndReaps(t *testing.T) {
 	require.NotNil(t, p.tmpls.updatedMeta.Prefetch, "persisted metadata must carry the mapping")
 	require.Equal(t, 2, p.tmpls.updatedMeta.Prefetch.Memory.Count(), "the persisted mapping must hold the harvested blocks")
 	require.True(t, p.uploadCalled, "remote metadata must be re-uploaded")
+	require.Equal(t, 1, p.tmpls.pins, "the throwaway runs off the template, so it must be pinned")
+	require.Equal(t, p.tmpls.pins, p.tmpls.releases, "the template pin must be released")
 }
 
 // TestHarvestRun_ThrowawayIdentity checks the throwaway resume runtime: it is
@@ -312,6 +325,8 @@ func TestHarvestRun_CollectErrorStillReaps(t *testing.T) {
 	require.True(t, p.inst.stopped && p.inst.closed, "throwaway must be reaped even when collection fails")
 	require.True(t, p.released)
 	require.False(t, p.tmpls.updated)
+	require.Equal(t, p.tmpls.pins, p.tmpls.releases,
+		"a pin leaked on the error path keeps the template resident for the life of the process")
 }
 
 // TestHarvestRun_AcquireErrorSkips: if the start slot can't be acquired the