iop/packages/go/streamgate/ingress_snapshot.go

package streamgate

import (
	"errors"
	"sync"
	"sync/atomic"
)

var (
	// ErrIngressSnapshotLimitExceeded is returned when retained bytes exceed max bytes limit.
	ErrIngressSnapshotLimitExceeded = errors.New("streamgate: ingress snapshot limit exceeded")

	// ErrIngressSnapshotInvalidHandle is returned when a backing handle is unissued or invalid.
	ErrIngressSnapshotInvalidHandle = errors.New("streamgate: ingress snapshot invalid backing handle")

	// ErrIngressSnapshotAlreadyBuilt is returned when mutating a builder that has already built a snapshot.
	ErrIngressSnapshotAlreadyBuilt = errors.New("streamgate: ingress snapshot builder already built")

	// ErrIngressSnapshotDuplicateCanonical is returned when setting canonical source more than once.
	ErrIngressSnapshotDuplicateCanonical = errors.New("streamgate: ingress snapshot duplicate canonical")

	// ErrIngressSnapshotDuplicateTyped is returned when registering a typed view with a duplicate name.
	ErrIngressSnapshotDuplicateTyped = errors.New("streamgate: ingress snapshot duplicate typed view name")

	// ErrIngressSnapshotClosed is returned when accessing a snapshot or lease that has been closed.
	ErrIngressSnapshotClosed = errors.New("streamgate: ingress snapshot closed")

	// ErrIngressSnapshotZeroMaxBytes is returned when maxBytes parameter is zero.
	ErrIngressSnapshotZeroMaxBytes = errors.New("streamgate: ingress snapshot max bytes must be greater than zero")

	// ErrIngressSnapshotNilCanonical is returned when Build is called without a canonical handle set.
	ErrIngressSnapshotNilCanonical = errors.New("streamgate: ingress snapshot missing canonical handle")

	// ErrIngressSnapshotRebuildCancelled is returned when a rebuild reservation is cancelled before commit.
	ErrIngressSnapshotRebuildCancelled = errors.New("streamgate: ingress snapshot rebuild reservation cancelled")

	// ErrIngressSnapshotRebuildInvalidState is returned when operating on a released or invalid rebuild guard.
	ErrIngressSnapshotRebuildInvalidState = errors.New("streamgate: ingress snapshot rebuild guard invalid state")
)

type rebuildOverflowError struct{}

func (rebuildOverflowError) Error() string {
	return "streamgate: ingress snapshot rebuild allocation overflow"
}

func (rebuildOverflowError) Unwrap() error {
	return ErrIngressSnapshotLimitExceeded
}

// ErrIngressSnapshotRebuildOverflow is returned when temporary rebuild allocation exceeds peak limit.
var ErrIngressSnapshotRebuildOverflow error = rebuildOverflowError{}

// globalOwnerCounter is a process-wide monotonic counter for builder owner IDs.
var globalOwnerCounter uint64

// BackingHandle identifies a payload blob owned by a specific builder.
// It is opaque and unexported fields prevent forging handles across builders.
type BackingHandle struct {
	ownerID uint64
	id      uint64
}

// BackingHandleZero is the zero-value handle used as a sentinel.
var BackingHandleZero = BackingHandle{}

// isValid returns true when the handle was issued by a real builder.
func (h BackingHandle) isValid() bool {
	return h.ownerID != 0
}

// TypedSemanticView describes a named typed semantic view of a backing blob.
type TypedSemanticView struct {
	name   string
	handle BackingHandle
}

// Name returns the typed view name.
func (v TypedSemanticView) Name() string { return v.name }

// Handle returns the backing handle referencing the typed view payload.
func (v TypedSemanticView) Handle() BackingHandle { return v.handle }

// snapshotBacking is the builder-owned payload blob referenced by a BackingHandle.
type snapshotBacking struct {
	payload []byte
}

// IngressSnapshotBuilder accumulates canonical and typed backing blobs
// under a byte limit, then transfers ownership of the backing store to an
// IngressSnapshot without copying payloads (canonical single-copy).
type IngressSnapshotBuilder struct {
	mu            sync.Mutex
	ownerID       uint64
	maxBytes      int64
	retainedBytes int64
	backings      map[BackingHandle]*snapshotBacking
	canonicalHdl  BackingHandle
	typedViews    []TypedSemanticView
	typedNames    map[string]BackingHandle
	nextID        uint64
	builderClosed bool
	builderBuilt  bool
	terminalErr   error
}

// NewIngressSnapshotBuilder creates a builder with the given byte limit.
// maxBytes must be positive.
func NewIngressSnapshotBuilder(maxBytes int64) (*IngressSnapshotBuilder, error) {
	if maxBytes <= 0 {
		return nil, ErrIngressSnapshotZeroMaxBytes
	}
	oid := atomic.AddUint64(&globalOwnerCounter, 1)
	return &IngressSnapshotBuilder{
		ownerID:    oid,
		maxBytes:   maxBytes,
		backings:   make(map[BackingHandle]*snapshotBacking),
		typedNames: make(map[string]BackingHandle),
		nextID:     1,
	}, nil
}

// IssueBackingHandle copies data into the builder-owned backing store,
// checks the byte limit before storing, and returns a handle bound to this builder.
// On overflow it transitions to a terminal failed state and returns BackingHandleZero
// along with ErrIngressSnapshotLimitExceeded.
func (b *IngressSnapshotBuilder) IssueBackingHandle(data []byte) (BackingHandle, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if err := b.requireOpen("IssueBackingHandle"); err != nil {
		return BackingHandle{}, err
	}
	return b.issueBackingHandleLocked(data, true)
}

// IssueOwnedBackingHandle transfers ownership of data into the builder without
// copying it. On success the caller must not mutate or retain data beyond the
// snapshot lifecycle. On error ownership remains with the caller.
func (b *IngressSnapshotBuilder) IssueOwnedBackingHandle(data []byte) (BackingHandle, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if err := b.requireOpen("IssueOwnedBackingHandle"); err != nil {
		return BackingHandle{}, err
	}
	return b.issueBackingHandleLocked(data, false)
}

// issueBackingHandleLocked accepts one backing after the caller has checked
// builder state. Must be called while holding b.mu.
func (b *IngressSnapshotBuilder) issueBackingHandleLocked(data []byte, copyData bool) (BackingHandle, error) {
	size := int64(len(data))
	if size > b.maxBytes-b.retainedBytes {
		return BackingHandle{}, b.failLocked(ErrIngressSnapshotLimitExceeded)
	}

	hdl := BackingHandle{ownerID: b.ownerID, id: b.nextID}
	b.nextID++

	payload := data
	if copyData {
		payload = append([]byte(nil), data...)
	}
	b.backings[hdl] = &snapshotBacking{payload: payload}
	b.retainedBytes += size

	return hdl, nil
}

// SetCanonicalWithHandle associates the given builder-owned backing handle
// as the canonical source. The payload is not re-accepted; only the handle
// is linked to the canonical slot.
func (b *IngressSnapshotBuilder) SetCanonicalWithHandle(hdl BackingHandle) error {
	b.mu.Lock()
	defer b.mu.Unlock()

	if err := b.requireOpen("SetCanonicalWithHandle"); err != nil {
		return err
	}
	if !b.validateHandle(hdl) {
		return ErrIngressSnapshotInvalidHandle
	}
	if b.canonicalHdl.isValid() {
		return ErrIngressSnapshotDuplicateCanonical
	}

	b.canonicalHdl = hdl
	return nil
}

// AddTypedView registers a named typed semantic view that references a
// builder-owned backing handle. The payload is not re-accepted; only the
// handle is linked to the typed view.
func (b *IngressSnapshotBuilder) AddTypedView(name string, hdl BackingHandle) error {
	b.mu.Lock()
	defer b.mu.Unlock()

	if err := b.requireOpen("AddTypedView"); err != nil {
		return err
	}
	if name == "" {
		return errors.New("streamgate: typed view name is required")
	}
	if !b.validateHandle(hdl) {
		return ErrIngressSnapshotInvalidHandle
	}
	if _, dup := b.typedNames[name]; dup {
		return ErrIngressSnapshotDuplicateTyped
	}

	b.typedViews = append(b.typedViews, TypedSemanticView{name: name, handle: hdl})
	b.typedNames[name] = hdl
	return nil
}

// Build moves the backing store and metadata into a new IngressSnapshot
// without copying payloads. The builder is poisoned into a built state
// and all subsequent mutations return ErrIngressSnapshotAlreadyBuilt.
func (b *IngressSnapshotBuilder) Build() (*IngressSnapshot, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if err := b.requireOpen("Build"); err != nil {
		return nil, err
	}
	if !b.validateHandle(b.canonicalHdl) {
		return nil, ErrIngressSnapshotNilCanonical
	}

	snap := &IngressSnapshot{
		maxBytes:   b.maxBytes,
		backings:   b.backings,
		canonical:  b.canonicalHdl,
		typedViews: b.typedViews,
	}

	b.backings = nil
	b.typedViews = nil
	b.typedNames = nil
	b.canonicalHdl = BackingHandle{}
	b.retainedBytes = 0
	b.builderBuilt = true

	return snap, nil
}

// Close idempotently releases builder-owned backing memory.
// Already-built or already-closed builders are no-ops.
// A failed (terminal) builder releases memory but does not clear the error state.
func (b *IngressSnapshotBuilder) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.builderClosed || b.builderBuilt {
		return
	}
	for _, sb := range b.backings {
		sb.payload = nil
	}
	b.backings = nil
	b.typedViews = nil
	b.typedNames = nil
	b.canonicalHdl = BackingHandle{}
	b.retainedBytes = 0
	b.builderClosed = true
}

// requireOpen returns an error if the builder cannot accept the named operation.
// Priority: built > terminal overflow > closed.
func (b *IngressSnapshotBuilder) requireOpen(op string) error {
	if b.builderBuilt {
		return ErrIngressSnapshotAlreadyBuilt
	}
	if b.terminalErr != nil {
		return b.terminalErr
	}
	if b.builderClosed {
		return ErrIngressSnapshotClosed
	}
	return nil
}

// validateHandle returns true when the handle was issued by this builder.
func (b *IngressSnapshotBuilder) validateHandle(hdl BackingHandle) bool {
	return hdl.isValid() && hdl.ownerID == b.ownerID && b.backings[hdl] != nil
}

// failLocked transitions the builder to a terminal failed state and releases
// all retained references. Must be called while holding b.mu.
func (b *IngressSnapshotBuilder) failLocked(err error) error {
	b.terminalErr = err
	b.builderClosed = true
	for _, sb := range b.backings {
		sb.payload = nil
	}
	b.backings = nil
	b.typedViews = nil
	b.typedNames = nil
	b.canonicalHdl = BackingHandle{}
	b.retainedBytes = 0
	return err
}

// fitsIngressSnapshotPeak returns true when retained+reserved+expected fits in maxBytes
// without int64 overflow. Uses subtraction-based checks so that any non-fitting
// combination (including near-MaxInt64 inputs) is rejected deterministically.
func fitsIngressSnapshotPeak(maxBytes, retained, reserved, expected int64) bool {
	if retained < 0 || reserved < 0 || expected < 0 {
		return false
	}
	remaining := maxBytes - retained
	if remaining < 0 {
		return false
	}
	if reserved > remaining {
		return false
	}
	remaining -= reserved
	if expected > remaining {
		return false
	}
	return true
}

// IngressSnapshot is the immutable result of a successful Build.
// It owns the backing store and provides defensive-copy accessors and rebuild guards.
type IngressSnapshot struct {
	mu                sync.RWMutex
	maxBytes          int64
	reservedTempBytes int64
	backings          map[BackingHandle]*snapshotBacking
	canonical         BackingHandle
	typedViews        []TypedSemanticView
	closed            bool
	activeGuards      []*IngressSnapshotRebuildGuard
}

// Accessor returns a read-only accessor for the snapshot.
func (s *IngressSnapshot) Accessor() IngressSnapshotAccessor {
	return IngressSnapshotAccessor{snap: s}
}

// Close closes the snapshot idempotently, fans out to all active rebuild guards,
// and releases backing references.
func (s *IngressSnapshot) Close() {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return
	}
	s.closeActiveGuardsLocked(nil)
	s.closeLocked()
}

// closeLocked releases backing references, marks the snapshot closed,
// and clears the active guard registry. Must be called while holding s.mu.
func (s *IngressSnapshot) closeLocked() {
	s.closed = true
	s.backings = nil
	s.reservedTempBytes = 0
	s.activeGuards = nil
}

// closeActiveGuardsLocked terminates all active rebuild guards except the
// optionally excluded current guard. For each sibling, if it holds a committed
// rebuilt snapshot the snapshot is closed, then the guard is released.
// Must be called while holding s.mu.
func (s *IngressSnapshot) closeActiveGuardsLocked(current *IngressSnapshotRebuildGuard) {
	guards := make([]*IngressSnapshotRebuildGuard, len(s.activeGuards))
	copy(guards, s.activeGuards)
	for _, g := range guards {
		if g == current {
			continue
		}
		g.mu.Lock()
		if !g.released {
			if g.state == RebuildStateCommitted && g.rebuiltSnapshot != nil {
				g.rebuiltSnapshot.Close()
			}
			g.releaseLocked(RebuildStateReleased)
		}
		g.mu.Unlock()
	}
}

// MaxBytes returns the maximum byte limit for this snapshot.
func (s *IngressSnapshot) MaxBytes() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.maxBytes
}

// ReservedTempBytes returns the sum of active temporary rebuild reservations.
func (s *IngressSnapshot) ReservedTempBytes() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.reservedTempBytes
}

// retainedCount returns the number of unique backing blobs. Must hold s.mu (write or RLock).
func (s *IngressSnapshot) retainedCount() int {
	return len(s.backings)
}

// retainedBytes returns the sum of backing payload sizes. Must hold s.mu.
func (s *IngressSnapshot) retainedBytes() int64 {
	var total int64
	for _, sb := range s.backings {
		total += int64(len(sb.payload))
	}
	return total
}

// hasBacking returns true when the backing store references the handle.
func (s *IngressSnapshot) hasBacking(hdl BackingHandle) bool {
	_, ok := s.backings[hdl]
	return ok
}

// RebuildGuardState represents the lifecycle state of a rebuild allocation guard.
type RebuildGuardState int

const (
	RebuildStateReserved RebuildGuardState = iota
	RebuildStateCommitted
	RebuildStateCancelled
	RebuildStateOverflow
	RebuildStateReleased
)

func (st RebuildGuardState) String() string {
	switch st {
	case RebuildStateReserved:
		return "reserved"
	case RebuildStateCommitted:
		return "committed"
	case RebuildStateCancelled:
		return "cancelled"
	case RebuildStateOverflow:
		return "overflow"
	case RebuildStateReleased:
		return "released"
	default:
		return "unknown"
	}
}

// IngressSnapshotRebuildGuard coordinates temporary rebuild output memory reservations,
// two-phase pre/post peak byte checks, and canonical/typed/temp reference release lifecycle.
type IngressSnapshotRebuildGuard struct {
	mu                sync.Mutex
	snap              *IngressSnapshot
	expectedTempBytes int64
	reservedBytes     int64
	tempPayload       []byte
	tempHandle        BackingHandle
	canonicalRef      BackingHandle
	typedRefs         []TypedSemanticView
	state             RebuildGuardState
	released          bool
	rebuiltSnapshot   *IngressSnapshot
}

// ReserveRebuild creates a two-phase rebuild allocation guard on the snapshot after checking
// pre-peak memory limits against maxBytes.
func (s *IngressSnapshot) ReserveRebuild(expectedTempBytes int64) (*IngressSnapshotRebuildGuard, error) {
	if expectedTempBytes < 0 {
		return nil, errors.New("streamgate: expected temporary bytes cannot be negative")
	}

	s.mu.Lock()
	defer s.mu.Unlock()

	if s.closed {
		return nil, ErrIngressSnapshotClosed
	}

	retained := s.retainedBytes()
	if !fitsIngressSnapshotPeak(s.maxBytes, retained, s.reservedTempBytes, expectedTempBytes) {
		return nil, ErrIngressSnapshotRebuildOverflow
	}

	s.reservedTempBytes += expectedTempBytes

	guard := &IngressSnapshotRebuildGuard{
		snap:              s,
		expectedTempBytes: expectedTempBytes,
		reservedBytes:     expectedTempBytes,
		canonicalRef:      s.canonical,
		typedRefs:         append([]TypedSemanticView(nil), s.typedViews...),
		state:             RebuildStateReserved,
	}

	s.activeGuards = append(s.activeGuards, guard)

	return guard, nil
}

// Commit completes the rebuild allocation by accepting actual output payload,
// checking post-peak memory limits, and returning a dispatchable IngressSnapshot.
func (g *IngressSnapshotRebuildGuard) Commit(actualOutput []byte) (*IngressSnapshot, error) {
	return g.CommitTyped("", actualOutput)
}

// CommitTyped completes the rebuild allocation with an optional typed view name for the output.
func (g *IngressSnapshotRebuildGuard) CommitTyped(typedName string, actualOutput []byte) (*IngressSnapshot, error) {
	return g.commitTyped(typedName, actualOutput, true)
}

// CommitOwnedTyped transfers ownership of actualOutput into the committed
// snapshot without copying it. On success the caller must not mutate or retain
// actualOutput beyond the returned snapshot lifecycle. On error ownership
// remains with the caller.
func (g *IngressSnapshotRebuildGuard) CommitOwnedTyped(typedName string, actualOutput []byte) (*IngressSnapshot, error) {
	return g.commitTyped(typedName, actualOutput, false)
}

func (g *IngressSnapshotRebuildGuard) commitTyped(typedName string, actualOutput []byte, copyOutput bool) (*IngressSnapshot, error) {
	if g == nil {
		return nil, ErrIngressSnapshotRebuildInvalidState
	}

	g.snap.mu.Lock()
	g.mu.Lock()
	defer g.mu.Unlock()
	defer g.snap.mu.Unlock()

	// Guard state check first for stable errors on terminal guards.
	if g.released || g.state != RebuildStateReserved {
		switch g.state {
		case RebuildStateOverflow:
			return nil, ErrIngressSnapshotRebuildOverflow
		case RebuildStateCancelled:
			return nil, ErrIngressSnapshotRebuildCancelled
		case RebuildStateCommitted:
			return nil, errors.New("streamgate: rebuild reservation already committed")
		case RebuildStateReleased:
			return nil, ErrIngressSnapshotClosed
		default:
			return nil, ErrIngressSnapshotRebuildInvalidState
		}
	}

	if g.snap.closed {
		g.releaseLocked(RebuildStateReleased)
		return nil, ErrIngressSnapshotClosed
	}

	actualSize := int64(len(actualOutput))
	retained := g.snap.retainedBytes()
	otherReserved := g.snap.reservedTempBytes - g.reservedBytes
	if otherReserved < 0 {
		otherReserved = 0
	}
	if !fitsIngressSnapshotPeak(g.snap.maxBytes, retained, otherReserved, actualSize) {
		// Post-peak overflow: release current guard, fan out to all active siblings,
		// close parent snapshot.
		g.releaseLocked(RebuildStateOverflow)
		g.snap.closeActiveGuardsLocked(g)
		g.snap.closeLocked()
		return nil, ErrIngressSnapshotRebuildOverflow
	}

	// Issue new backing handle for temporary rebuild output
	oid := atomic.AddUint64(&globalOwnerCounter, 1)
	newHdl := BackingHandle{ownerID: oid, id: 1}
	payload := actualOutput
	if copyOutput {
		payload = append([]byte(nil), actualOutput...)
	}
	sb := &snapshotBacking{payload: payload}

	// Build new backings map
	newBackings := make(map[BackingHandle]*snapshotBacking, len(g.snap.backings)+1)
	for h, b := range g.snap.backings {
		newBackings[h] = b
	}
	newBackings[newHdl] = sb

	// Build new typed views list
	newTypedViews := make([]TypedSemanticView, 0, len(g.snap.typedViews)+1)
	newTypedViews = append(newTypedViews, g.snap.typedViews...)
	if typedName != "" {
		newTypedViews = append(newTypedViews, TypedSemanticView{name: typedName, handle: newHdl})
	}

	rebuilt := &IngressSnapshot{
		maxBytes:   g.snap.maxBytes,
		backings:   newBackings,
		canonical:  g.snap.canonical,
		typedViews: newTypedViews,
	}

	// Unreserve temporary bytes from parent snapshot
	g.snap.reservedTempBytes -= g.reservedBytes
	if g.snap.reservedTempBytes < 0 {
		g.snap.reservedTempBytes = 0
	}
	g.reservedBytes = 0

	g.tempPayload = sb.payload
	g.tempHandle = newHdl
	g.rebuiltSnapshot = rebuilt
	g.state = RebuildStateCommitted

	return rebuilt, nil
}

// Cancel releases the rebuild reservation and drops references in defined order (temp -> canonical -> typed).
func (g *IngressSnapshotRebuildGuard) Cancel() error {
	if g == nil {
		return nil
	}

	g.snap.mu.Lock()
	g.mu.Lock()
	defer g.mu.Unlock()
	defer g.snap.mu.Unlock()

	if g.released {
		return nil
	}

	g.releaseLocked(RebuildStateCancelled)
	return nil
}

// Close idempotently releases the rebuild guard resources.
func (g *IngressSnapshotRebuildGuard) Close() {
	if g == nil {
		return
	}

	g.snap.mu.Lock()
	g.mu.Lock()
	defer g.mu.Unlock()
	defer g.snap.mu.Unlock()

	if g.released {
		return
	}

	g.releaseLocked(RebuildStateReleased)
}

// releaseLocked releases references in defined order: temporary -> canonical -> typed.
// Must be called while holding both g.snap.mu and g.mu.
func (g *IngressSnapshotRebuildGuard) releaseLocked(st RebuildGuardState) {
	if g.released {
		return
	}
	g.state = st
	g.released = true

	// Step 1: Release temporary reference
	g.tempPayload = nil
	g.tempHandle = BackingHandle{}

	// Step 2: Release canonical reference
	g.canonicalRef = BackingHandle{}

	// Step 3: Release typed references
	g.typedRefs = nil

	// Step 4: Release rebuilt snapshot reference (committed path)
	g.rebuiltSnapshot = nil

	// Unreserve temporary bytes from parent snapshot
	if g.reservedBytes > 0 && g.snap != nil {
		g.snap.reservedTempBytes -= g.reservedBytes
		if g.snap.reservedTempBytes < 0 {
			g.snap.reservedTempBytes = 0
		}
		g.reservedBytes = 0
	}

	// Unregister from parent's active guard registry
	if g.snap != nil {
		snap := g.snap
		for i, ag := range snap.activeGuards {
			if ag == g {
				snap.activeGuards = append(snap.activeGuards[:i], snap.activeGuards[i+1:]...)
				break
			}
		}
	}
}

// State returns the current lifecycle state of the guard.
func (g *IngressSnapshotRebuildGuard) State() RebuildGuardState {
	if g == nil {
		return RebuildStateReleased
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.state
}

// IsReleased returns true if the guard has been released or cancelled.
func (g *IngressSnapshotRebuildGuard) IsReleased() bool {
	if g == nil {
		return true
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.released
}

// ReservedBytes returns the number of expected temporary bytes currently reserved.
func (g *IngressSnapshotRebuildGuard) ReservedBytes() int64 {
	if g == nil {
		return 0
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.reservedBytes
}

// RebuiltSnapshot returns the committed rebuilt IngressSnapshot held by this guard,
// or nil if the guard has not committed successfully (overflow, cancelled, released).
func (g *IngressSnapshotRebuildGuard) RebuiltSnapshot() *IngressSnapshot {
	if g == nil {
		return nil
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.rebuiltSnapshot
}

// DispatchableSnapshot returns the committed rebuilt IngressSnapshot, or an error if not committed or released.
func (g *IngressSnapshotRebuildGuard) DispatchableSnapshot() (*IngressSnapshot, error) {
	if g == nil {
		return nil, ErrIngressSnapshotRebuildInvalidState
	}
	g.mu.Lock()
	defer g.mu.Unlock()

	if g.state == RebuildStateCommitted && g.rebuiltSnapshot != nil {
		return g.rebuiltSnapshot, nil
	}
	switch g.state {
	case RebuildStateOverflow:
		return nil, ErrIngressSnapshotRebuildOverflow
	case RebuildStateCancelled:
		return nil, ErrIngressSnapshotRebuildCancelled
	case RebuildStateReleased:
		return nil, ErrIngressSnapshotClosed
	default:
		return nil, ErrIngressSnapshotRebuildInvalidState
	}
}

// Accessor returns an IngressSnapshotAccessor for the committed snapshot, or a closed accessor if released.
func (g *IngressSnapshotRebuildGuard) Accessor() IngressSnapshotAccessor {
	snap, err := g.DispatchableSnapshot()
	if err != nil {
		return IngressSnapshotAccessor{snap: &IngressSnapshot{closed: true}}
	}
	return snap.Accessor()
}

// HasTempReference returns true if the temporary handle/payload reference is active.
func (g *IngressSnapshotRebuildGuard) HasTempReference() bool {
	if g == nil {
		return false
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.tempHandle.isValid() || len(g.tempPayload) > 0
}

// HasCanonicalReference returns true if the canonical reference is active.
func (g *IngressSnapshotRebuildGuard) HasCanonicalReference() bool {
	if g == nil {
		return false
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.canonicalRef.isValid()
}

// HasTypedReferences returns true if typed references are active.
func (g *IngressSnapshotRebuildGuard) HasTypedReferences() bool {
	if g == nil {
		return false
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return len(g.typedRefs) > 0
}

// IngressSnapshotAccessor provides read-only, defensive-copy access to an IngressSnapshot.
type IngressSnapshotAccessor struct {
	snap *IngressSnapshot
}

// MaxBytes returns the maximum byte limit configured for the snapshot.
func (a IngressSnapshotAccessor) MaxBytes() int64 {
	return a.snap.MaxBytes()
}

// ReservedTempBytes returns the active temporary reserved bytes on the snapshot.
func (a IngressSnapshotAccessor) ReservedTempBytes() int64 {
	return a.snap.ReservedTempBytes()
}

// IsClosed returns true if the snapshot has been closed.
func (a IngressSnapshotAccessor) IsClosed() bool {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	return a.snap.closed
}

// RetainedCount returns the number of unique backing blobs.
func (a IngressSnapshotAccessor) RetainedCount() int {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	return a.snap.retainedCount()
}

// RetainedBytes returns the sum of backing payload sizes.
func (a IngressSnapshotAccessor) RetainedBytes() int64 {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	return a.snap.retainedBytes()
}

// HasCanonical returns true if a canonical handle was set during build.
func (a IngressSnapshotAccessor) HasCanonical() bool {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	return a.snap.canonical.isValid()
}

// Canonical returns a defensive copy of the canonical payload, or an error.
func (a IngressSnapshotAccessor) Canonical() ([]byte, error) {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	if a.snap.closed {
		return nil, ErrIngressSnapshotClosed
	}
	if !a.snap.canonical.isValid() {
		return nil, ErrIngressSnapshotInvalidHandle
	}
	sb, ok := a.snap.backings[a.snap.canonical]
	if !ok {
		return nil, ErrIngressSnapshotInvalidHandle
	}
	out := make([]byte, len(sb.payload))
	copy(out, sb.payload)
	return out, nil
}

// CanonicalAlias returns a read-only alias of the canonical backing. The alias
// is valid only while the snapshot is open and must never be mutated. Callers
// that need independent ownership must use Canonical instead.
func (a IngressSnapshotAccessor) CanonicalAlias() ([]byte, error) {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	if a.snap.closed {
		return nil, ErrIngressSnapshotClosed
	}
	if !a.snap.canonical.isValid() {
		return nil, ErrIngressSnapshotInvalidHandle
	}
	sb, ok := a.snap.backings[a.snap.canonical]
	if !ok {
		return nil, ErrIngressSnapshotInvalidHandle
	}
	return sb.payload, nil
}

// TypedView returns a defensive copy of the typed view payload by name, or an error.
func (a IngressSnapshotAccessor) TypedView(name string) ([]byte, error) {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	if a.snap.closed {
		return nil, ErrIngressSnapshotClosed
	}
	for _, v := range a.snap.typedViews {
		if v.name == name {
			sb, ok := a.snap.backings[v.handle]
			if !ok {
				return nil, ErrIngressSnapshotInvalidHandle
			}
			out := make([]byte, len(sb.payload))
			copy(out, sb.payload)
			return out, nil
		}
	}
	return nil, ErrIngressSnapshotInvalidHandle
}

// TypedViewAlias returns a read-only alias of a typed-view backing. The alias
// is valid only while the snapshot is open and must never be mutated. Callers
// that need independent ownership must use TypedView instead.
func (a IngressSnapshotAccessor) TypedViewAlias(name string) ([]byte, error) {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	if a.snap.closed {
		return nil, ErrIngressSnapshotClosed
	}
	for _, v := range a.snap.typedViews {
		if v.name == name {
			sb, ok := a.snap.backings[v.handle]
			if !ok {
				return nil, ErrIngressSnapshotInvalidHandle
			}
			return sb.payload, nil
		}
	}
	return nil, ErrIngressSnapshotInvalidHandle
}

// TypedViewNames returns the names of all typed views in registration order.
func (a IngressSnapshotAccessor) TypedViewNames() []string {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	out := make([]string, 0, len(a.snap.typedViews))
	for _, v := range a.snap.typedViews {
		out = append(out, v.name)
	}
	return out
}

// RetainedBackings returns the set of backing handles retained by the snapshot.
func (a IngressSnapshotAccessor) RetainedBackings() []BackingHandle {
	a.snap.mu.RLock()
	defer a.snap.mu.RUnlock()
	out := make([]BackingHandle, 0, len(a.snap.backings))
	for hdl := range a.snap.backings {
		out = append(out, hdl)
	}
	return out
}