feat(orch): add memfd identity-offset source for provisional-header serving

Add DedupedMemfdCache.ServeMemfd (identity device-offset reads guarded by the
existing memfd RWMutex) and MemfdIdentitySource, a local diff source that serves
dirty pages from the still-mapped memfd while dedup runs. This is what a
provisional (pre-dedup) memfile header reads from, so a resume overlapping the
pause can serve dirty pages immediately instead of waiting for the deduped
header. Publish the memfd at construction and free it exactly once under the
write lock (releaseMemfd) so serving can't race the drain's memfd close. Composes
with the in-flight drain serving rather than replacing it. Not yet wired into the
pause/resume path.

Signed-off-by: Nikita Kalyazin <nikita.kalyazin@e2b.dev>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: kalyazin

packages/orchestrator/pkg/sandbox/block/memfd.go

@@ -275,6 +275,11 @@ func NewCacheFromMemfdDeduped(
 		cancel:   cancel,
 		done:     utils.NewSetOnce[*Cache](),
 		inflight: inflightServe,
+		// Publish the memfd up front (guarded by mu) so a provisional-header
+		// resume can serve dirty pages via ServeMemfd during the compare window,
+		// before metaOut/the deduped header resolves. runDedup frees it under mu
+		// after the drain.
+		memfd: memfd,
 	}
 	go d.runDedup(drainCtx, base, blockSize, memfd, dirty, bestEffort, directIO, budget, inputEmpty, metaOut)
 
@@ -353,7 +358,7 @@ func (d *DedupedMemfdCache) runDedup(
 	compareDur := time.Since(compareStart)
 	if err != nil {
 		logSetOnceErr(ctx, "dedup metaOut", metaOut.SetError(err))
-		logSetOnceErr(ctx, "dedup done", d.done.SetError(errors.Join(err, memfd.Close())))
+		logSetOnceErr(ctx, "dedup done", d.done.SetError(errors.Join(err, d.releaseMemfd())))
 
 		return
 	}
@@ -388,11 +393,7 @@ func (d *DedupedMemfdCache) runDedup(
 	writeStart := time.Now()
 	cache, err := dedupDrain(ctx, src, plan.pageDirty, blockSize, d.outPath, directIO)
 	writeDur := time.Since(writeStart)
-	d.mu.Lock()
-	closeErr := memfd.Close()
-	d.memfd = nil
-	d.mu.Unlock()
-	if closeErr != nil {
+	if closeErr := d.releaseMemfd(); closeErr != nil {
 		logger.L().Warn(ctx, "close memfd after dedup drain", zap.Error(closeErr))
 	}
 
@@ -475,6 +476,40 @@ func (d *DedupedMemfdCache) Slice(off, length int64) ([]byte, error) {
 	return c.Slice(off, length)
 }
 
+// releaseMemfd closes the memfd exactly once, under the write lock so it can't
+// be unmapped beneath an in-flight ServeMemfd/tryInflightRead reader.
+func (d *DedupedMemfdCache) releaseMemfd() error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.memfd == nil {
+		return nil
+	}
+	err := d.memfd.Close()
+	d.memfd = nil
+
+	return err
+}
+
+// ServeMemfd copies [off, off+len(b)) from the still-mapped memfd using
+// identity (device) addressing, for a provisional local header that attributes
+// dirty pages to the memfd at their device offset. Returns BytesNotAvailableError
+// once the memfd has been released (drain finished), so the caller falls back to
+// the drained cache via the swapped-in deduped header.
+func (d *DedupedMemfdCache) ServeMemfd(b []byte, off int64) (int, error) {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	if d.memfd == nil {
+		return 0, BytesNotAvailableError{}
+	}
+	src, err := d.memfd.Slice(off, int64(len(b)))
+	if err != nil {
+		return 0, err
+	}
+	copy(b, src)
+
+	return len(b), nil
+}
+
 func (d *DedupedMemfdCache) Close() error {
 	d.cancel()
 	c, _ := d.done.Wait()
@@ -486,6 +521,51 @@ func (d *DedupedMemfdCache) Close() error {
 	return nil
 }
 
+// MemfdIdentitySource is the provisional local diff source: it serves dirty
+// pages from a DedupedMemfdCache's still-mapped memfd at identity (device)
+// offsets while dedup runs. It backs a distinct provisional build id; once the
+// deduped header is swapped in, reads route to the real build id instead and
+// this source is dropped. It is never uploaded.
+type MemfdIdentitySource struct {
+	d    *DedupedMemfdCache
+	size int64
+}
+
+func NewMemfdIdentitySource(d *DedupedMemfdCache, size int64) *MemfdIdentitySource {
+	return &MemfdIdentitySource{d: d, size: size}
+}
+
+func (s *MemfdIdentitySource) ReadAt(b []byte, off int64) (int, error) { return s.d.ServeMemfd(b, off) }
+
+func (s *MemfdIdentitySource) Slice(off, length int64) ([]byte, error) {
+	b := make([]byte, length)
+	if _, err := s.d.ServeMemfd(b, off); err != nil {
+		return nil, err
+	}
+
+	return b, nil
+}
+
+// IsCached reports the provisional source as resident while the memfd is mapped,
+// so dedup best-effort peeks and build.File.IsCached treat it as local.
+func (s *MemfdIdentitySource) IsCached(_ context.Context, off, length int64) bool {
+	s.d.mu.RLock()
+	defer s.d.mu.RUnlock()
+
+	return s.d.memfd != nil && off >= 0 && off+length <= s.size
+}
+
+func (s *MemfdIdentitySource) Size() (int64, error)                    { return s.size, nil }
+func (s *MemfdIdentitySource) BlockSize() int64                        { return header.PageSize }
+func (s *MemfdIdentitySource) Close() error                            { return nil }
+func (s *MemfdIdentitySource) FileSize(context.Context) (int64, error) { return s.size, nil }
+
+// Path has no meaning for a memfd-backed source; it is local-only and never
+// uploaded, so the upload path (which needs a file path) must never reach it.
+func (s *MemfdIdentitySource) Path(context.Context) (string, error) {
+	return "", errors.New("provisional memfd source has no path")
+}
+
 func (d *DedupedMemfdCache) Path(ctx context.Context) (string, error) {
 	c, err := d.Wait(ctx)
 	if err != nil {