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 }