IRollupArchiveRuntimeStore
Namespace: Meshmakers.Octo.Runtime.Contracts.StreamData
Abstracts the persistence operations on CkRollupArchive entities required by the rollup
orchestrator and the lifecycle service. Decouples those services from any particular runtime
repository implementation (Mongo today, possibly others later) and from the generated
CkRollupArchive class itself, so the orchestrator stays in Runtime.Engine without
taking a hard dependency on the StreamData CK model package.
public interface IRollupArchiveRuntimeStore
Remarks:
Mirrors IArchiveRuntimeStore for the rollup-specific entity. Implementations must reject schema-relevant mutations once the rollup has left CkArchiveStatus.Created (concept §7) and must maintain RollupArchiveSnapshot.LastAggregatedBucketEnd and RollupArchiveSnapshot.FrozenUntil as monotonic where the concept requires it (§6).
Methods
GetAsync(OctoObjectId)
Reads the current state of the rollup identified by rollupRtId, or
null if no such entity exists (or has been soft-deleted).
Task<RollupArchiveSnapshot> GetAsync(OctoObjectId rollupRtId)
Parameters
rollupRtId OctoObjectId
Returns
InsertAsync(String, RtCkId<CkTypeId>, IReadOnlyList<RollupSourceReference>, TimeSpan, TimeSpan, IReadOnlyList<CkRollupAggregationSpec>, IReadOnlyList<CkArchiveColumnSpec>, BucketAlignment, String, Nullable<TimeSpan>)
Inserts a new CkRollupArchive entity in CkArchiveStatus.Created. The shared
archive lifecycle service handles status transitions afterwards (Activate, Disable, etc.);
this method exists separately so the lifecycle service can derive the inherited CkArchive
attributes (targetCkTypeId, columns) from the source
archive and the aggregations, keeping the generic CkEntity GraphQL mutation pipeline free
of rollup-specific knowledge. Concept §4, §9.
Task<OctoObjectId> InsertAsync(string rtWellKnownName, RtCkId<CkTypeId> targetCkTypeId, IReadOnlyList<RollupSourceReference> sources, TimeSpan bucketSize, TimeSpan watermarkLag, IReadOnlyList<CkRollupAggregationSpec> aggregations, IReadOnlyList<CkArchiveColumnSpec> columns, BucketAlignment bucketAlignment, string referenceTimeZone, Nullable<TimeSpan> carryLookback)
Parameters
rtWellKnownName String
Optional human-readable name. Null falls back to the rtId.
targetCkTypeId RtCkId<CkTypeId>
CK type the rollup rows live on — inherited from the source archives.
sources IReadOnlyList<RollupSourceReference>
The source archives (CkArchive, or CkRollupArchive for chained rollups) this rollup
aggregates from, each with an optional validity span (AB#5157). Implementations persist
this list on the Sources attribute only; the deprecated
SourceArchiveRtId scalar is never written, not even for a single unbounded source.
bucketSize TimeSpan
Bucket width.
watermarkLag TimeSpan
How long the orchestrator waits after bucket-end before aggregating.
aggregations IReadOnlyList<CkRollupAggregationSpec>
User-defined aggregation specs.
columns IReadOnlyList<CkArchiveColumnSpec>
Derived storage columns (from RollupColumnGenerator.Generate(IReadOnlyList<CkRollupAggregationSpec>)). Stored on the
inherited Columns slot for mandatory-attribute validation; the read path re-derives from
aggregations to stay authoritative.
bucketAlignment BucketAlignment
Bucket-boundary alignment. Defaults to BucketAlignment.FixedSize for callers
(and pre-1.4.0 entities) that don't need calendar-aware buckets. Concept-time-range §7.
referenceTimeZone String
Optional IANA reference time-zone (e.g. Europe/Vienna) that aligns calendar bucket
boundaries to local wall-clock time so they are DST-correct. null keeps UTC calendar
boundaries. Only meaningful for calendar bucketAlignment variants; ignored
for BucketAlignment.FixedSize. AB#4300 / decision O6.
carryLookback Nullable<TimeSpan>
Optional bound on the TimeWeightedAvg carry-in scan (LOCF opening state), persisted as
CarryLookbackMs. null ⇒ attribute unset ⇒ the engine default of 35 days.
AB#4336 / decision D1.
Returns
Task<OctoObjectId>
The generated runtime id of the new rollup archive.
ArchiveEntityAsync(OctoObjectId)
Soft-deletes the rollup entity by setting rtState = Archived. The Crate table is
dropped separately by the lifecycle service via IStreamDataRepository.
Task ArchiveEntityAsync(OctoObjectId rollupRtId)
Parameters
rollupRtId OctoObjectId
Returns
AdvanceWatermarkAsync(OctoObjectId, DateTime, Boolean)
Advances RollupArchiveSnapshot.LastAggregatedBucketEnd to
bucketEnd. Called by the orchestrator immediately after a bucket's rows
have been upserted into the rollup table. Implementations should reject backwards moves
unless allowRewind is true (used by the rewindRollupWatermark
mutation).
Task AdvanceWatermarkAsync(OctoObjectId rollupRtId, DateTime bucketEnd, bool allowRewind)
Parameters
rollupRtId OctoObjectId
bucketEnd DateTime
allowRewind Boolean
Returns
SetFrozenUntilAsync(OctoObjectId, Nullable<DateTime>)
Sets RollupArchiveSnapshot.FrozenUntil. Implementations enforce monotonicity:
the new value must be greater than or equal to the current one, except when the
unfreezeRollupArchive mutation is called (which passes null). Concept §6.
Task SetFrozenUntilAsync(OctoObjectId rollupRtId, Nullable<DateTime> frozenUntil)
Parameters
rollupRtId OctoObjectId
frozenUntil Nullable<DateTime>
Returns
EnumerateAsync()
Enumerates every non-soft-deleted CkRollupArchive entity in the tenant. Used by the rollup orchestrator background worker on each tick to pick up rollups that are due for processing. Order is implementation-defined; callers must not rely on it.
IAsyncEnumerable<RollupArchiveSnapshot> EnumerateAsync()
Returns
IAsyncEnumerable<RollupArchiveSnapshot>
TryPersistDerivedColumnsAsync(OctoObjectId)
Persists the derived aggregate columns (RollupColumnGenerator.Generate(IReadOnlyList<CkRollupAggregationSpec>) over
the entity's aggregations) onto the inherited Archive.Columns attribute when no
persisted (non-computed) column is present yet, preserving any computed columns the entity
already declares. Self-heal for rollups created outside the API path (e.g. ImportRt seeds)
whose missing mandatory Columns attribute breaks the non-null columns GraphQL field
(AB#4771/AB#4772). Idempotent; a no-op on healthy entities.
Task<bool> TryPersistDerivedColumnsAsync(OctoObjectId rollupRtId)
Parameters
rollupRtId OctoObjectId
Returns
Task<Boolean>
True when columns were written; false when nothing needed healing (or the RtId is
not a live rollup).
CountActiveRollupsForSourceAsync(OctoObjectId)
Returns the count of non-soft-deleted rollups that list
sourceArchiveRtId among their sources — in either storage form, and
regardless of the validity span the reference carries (AB#5157). Used by the source
archive's delete path to enforce RollupSourceInUseException (concept §6).
Task<int> CountActiveRollupsForSourceAsync(OctoObjectId sourceArchiveRtId)
Parameters
sourceArchiveRtId OctoObjectId