Documentation
¶
Overview ¶
Package reflow contains embeddable workflow substrate for transfer reflow.
The substrate began (v0.3.4) with the adaptive concurrency contract, typed JSONL record/config payloads, and provider-error redaction helpers shared by the CLI and library runners. It now adds the embeddable engine contract: Runner (NewRunner/Run), Config, the typed Source forms (ObjectSource, PrefixSource, FileTreeSource, RecordStreamSource), the structured Destination, the EventSink event boundary, and the minimal CheckpointStore resume interface.
The execution surface is migrating incrementally. Runner.Run executes the migrated paths — currently the dry-run and copy planes over a RecordStreamSource: it streams S3 reflow-input records, plans destination mappings, emits typed events, performs conditional PUTs for copied objects, and returns an InvalidInputsError after the summary when a stream carries invalid records (mirroring the command path's non-zero exit). The copy plane executes the skip-if-duplicate, fail, overwrite, and overwrite-if-source-newer collision modes; forms and scenarios not yet migrated (object/prefix/file-tree sources, the quarantine collision cell, and provenance sidecars) return ErrNotImplemented, decided from the source form and config before any stream bytes are read, so a caller can fall back to the CLI path with the same Source. Providers are injected as provider.Provider handles, so the engine never imports a concrete provider, SDK, or the storageful (sqlite/index-store) graph — enforced by the dependency-boundary test.
The package is Experimental; see docs/api-stability.md for the stability tiers.
Index ¶
- Constants
- Variables
- func ConcurrencyConnectionError(err error) bool
- func DestAdmittedN(cfg ConcurrencyConfig) int
- func FormatErrorMessage(prefix string, err error) string
- func IsNonAdvisoryFamily(recordType string) bool
- func IsRelativeQuarantinePrefix(prefix string) bool
- func IsValidPutStorageClass(storageClass string) bool
- func IsValidRecordTypeToken(raw string) bool
- func NewLimiterCopyGate(limiter *ConcurrencyLimiter, stats *StageStats) transfer.CopyGate
- func NewPathError(defaultMessage string, verboseMessage string, opts PathErrorOptions) error
- func ProvenanceNow() string
- func QuarantineDestRel(prefix string, sourceKey string) string
- func RequireSourceNewerCapability(dst provider.Provider) error
- func SafeReceiptHandle(value string) string
- func SafeReceiptSHA256(value string) string
- func SanitizeOperationCauseMessage(err error) string
- func SanitizeRecordType(raw string) string
- func ValidateMetadataBudget(metadata map[string]string) error
- func ValidatePerObjectMetadataRules(sourceRules []MetadataSourceKeyRule, derivedRules []MetadataDerivedRule) error
- func ValidateProvenanceModeAndPolicy(mode, onWriteError string) error
- func ValidateProvenanceSuffix(suffix string, allowUnsafe bool) error
- type AdvisoryWarningTracker
- func (t *AdvisoryWarningTracker) CapSummaryWarning() (Warning, bool)
- func (t *AdvisoryWarningTracker) DistinctTypes() int
- func (t *AdvisoryWarningTracker) Observe(recordType string) (Warning, bool)
- func (t *AdvisoryWarningTracker) SkippedRecords() int64
- func (t *AdvisoryWarningTracker) UnreportedRecords() int64
- func (t *AdvisoryWarningTracker) UnreportedTypes() int
- type CheckpointCollision
- type CheckpointItem
- type CheckpointStore
- type CheckpointWriterStatsRecord
- type CollisionInfo
- type CollisionPolicy
- type ConcurrencyConfig
- type ConcurrencyLimiter
- func (l *ConcurrencyLimiter) Acquire(ctx context.Context) (func(), error)
- func (l *ConcurrencyLimiter) AcquireDest(ctx context.Context) (func(), error)
- func (l *ConcurrencyLimiter) AcquireSource(ctx context.Context) (func(), error)
- func (l *ConcurrencyLimiter) DestDomainCeiling() int
- func (l *ConcurrencyLimiter) ObserveConnectionError()
- func (l *ConcurrencyLimiter) ObserveProviderResult(err error)
- func (l *ConcurrencyLimiter) ObserveSuccess()
- func (l *ConcurrencyLimiter) ObserveThrottle()
- func (l *ConcurrencyLimiter) ReserveCopyMemory(ctx context.Context, sourceSize int64) (func(), error)
- func (l *ConcurrencyLimiter) ResetOccupancyWindow()
- func (l *ConcurrencyLimiter) RetryBufferCap() int64
- func (l *ConcurrencyLimiter) Snapshot() ConcurrencyStats
- type ConcurrencyStats
- type Config
- type DestLayout
- type Destination
- type EnvelopeInspection
- func (e EnvelopeInspection) FormatAdvisoryMalformedError() error
- func (e EnvelopeInspection) FormatCLIUnknownTypeError() error
- func (e EnvelopeInspection) FormatEngineUnknownTypeError() error
- func (e EnvelopeInspection) FormatFirstRecordUnknownTypeReason(malformedPrefix, unknownTypePrefix string) string
- func (e EnvelopeInspection) IsSkippableAdvisoryUnknown() bool
- type ErrorEvent
- type EventSink
- type FileTreeSource
- type IfAbsentCapability
- type InvalidInputsError
- type LaneCheckpointStore
- type MetadataBudgetError
- type MetadataDerivationError
- type MetadataDerivedRule
- type MetadataPlan
- func (c MetadataPlan) GoString() string
- func (c MetadataPlan) HasPerObjectRules() bool
- func (c MetadataPlan) NeedsSourceHead() bool
- func (c MetadataPlan) PutOptions(source *provider.ObjectMeta) (provider.PutOptions, error)
- func (c MetadataPlan) RequiresCapability() bool
- func (c MetadataPlan) String() string
- func (c MetadataPlan) Validate() error
- type MetadataRunConfig
- type MetadataSourceKeyRule
- type MissingConditionalCapabilityError
- type ObjectErrorsError
- type ObjectPathStageStatsEmitter
- type ObjectPathStageStatsRecord
- type ObjectSource
- type PathErrorOptions
- type PrefixSource
- type ProvenanceEmitter
- type ProvenancePlacementContext
- type ProvenancePlacementPlan
- type ProvenancePlan
- func (p ProvenancePlan) Enabled() bool
- func (p ProvenancePlan) GoString() string
- func (p ProvenancePlan) RunConfig() *ProvenanceRunConfig
- func (p ProvenancePlan) String() string
- func (p ProvenancePlan) Validate() error
- func (p ProvenancePlan) ValidateAgainstDestination(destProviderID, destBucket string) error
- type ProvenancePutFunc
- type ProvenanceRef
- type ProvenanceRunConfig
- type ProvenanceSidecarInput
- type ProvenanceSidecarPayload
- type ProvenanceSidecarRoot
- type Record
- type RecordStreamSource
- type ResourceProbe
- type RewriteConfig
- type RunRecord
- type Runner
- type Source
- type SourceEnumerationError
- type SourceMetadataCollisionError
- type SourceResolver
- type SourceRunMetadata
- type SourceRunMetadataStore
- type SourceRunRecord
- type StageStats
- type Summary
- type SummaryRecord
- type Warning
Constants ¶
const ( // AdvisoryRecordSkippedWarningCode is the warning code emitted (at most once // per distinct record type per run, up to MaxAdvisoryWarningTypes) when a // reflow consumer skips an unknown record envelope marked with top-level // advisory=true. AdvisoryRecordSkippedWarningCode = "REFLOW_ADVISORY_RECORD_SKIPPED" // AdvisoryWarningCapReachedCode is the aggregated warning code emitted once // at the end of a stream when more than MaxAdvisoryWarningTypes distinct // unknown advisory record types were skipped during the run. AdvisoryWarningCapReachedCode = "REFLOW_ADVISORY_WARNING_CAP_REACHED" // MaxAdvisoryWarningTypes bounds the number of distinct unknown advisory // record types that can emit per-type warnings in a single reflow run. MaxAdvisoryWarningTypes = 16 // MaxUnreportedAdvisoryTypesTracked bounds the number of distinct unreported // advisory record types tracked in memory after MaxAdvisoryWarningTypes is // reached. MaxUnreportedAdvisoryTypesTracked = 1024 )
const ( ErrCodeAccessDenied = "ACCESS_DENIED" ErrCodeNotFound = "NOT_FOUND" ErrCodeThrottled = "THROTTLED" ErrCodeTransient = "TRANSIENT" ErrCodeAlreadyExists = "ALREADY_EXISTS" ErrCodeInternal = "INTERNAL" )
const ( CollisionSkipIfDuplicate = "skip-if-duplicate" CollisionFail = "fail" CollisionOverwrite = "overwrite" CollisionQuarantine = "quarantine" CollisionOverwriteIfSourceNewer = "overwrite-if-source-newer" )
Collision mode values. These are the resolved --on-collision strings the engine acts on.
const ( MetadataPolicyClear = "clear" MetadataPolicyPreserve = "preserve" MetadataPolicyMerge = "merge" // MetadataOnMissing* selects how a per-object rule reacts when its source // value is missing or unrepresentable. MetadataOnMissingSkip = "skip" MetadataOnMissingFail = "fail" MetadataOnMissingEmpty = "empty" // MetadataStorageClassPropagate copies the source object's storage class to // the destination rather than setting a fixed class. MetadataStorageClassPropagate = "propagate" )
Destination user-metadata policy values. These are the resolved policy strings the engine acts on; the CLI maps its --metadata-policy flag onto them.
const ( ProvenanceSchema = "gonimbus.provenance.v1" ProvenanceSchemaVersion = "1.0.0" // ProvenanceWriteFailedWarningCode is the warning code emitted when a sidecar // write fails under the warn policy. ProvenanceWriteFailedWarningCode = "PROVENANCE_WRITE_FAILED" )
Provenance sidecar schema identity. These are the wire contract for the gonimbus.provenance.v1 audit object; do not change without a schema revision.
const ( ProvenanceOnWriteErrorWarn = "warn" ProvenanceOnWriteErrorFail = "fail" )
On-write-error policy values shared by both surfaces.
const ( ProvenanceModeNone = "none" ProvenanceModeSidecar = "sidecar" ProvenancePlacementSibling = "sibling" ProvenancePlacementMirror = "mirrored-root" )
Provenance mode and placement values shared by both surfaces.
const ( // RecordType is the JSONL type for per-object transfer reflow records. RecordType = "gonimbus.reflow.v1" // RunRecordType is the JSONL type for transfer reflow run configuration. RunRecordType = "gonimbus.reflow.run.v1" // SourceRecordType is the JSONL type for transfer reflow source metadata. SourceRecordType = "gonimbus.reflow.source.v1" // SummaryRecordType is the JSONL type for transfer reflow summaries. SummaryRecordType = "gonimbus.reflow.summary.v1" // CheckpointWriterStatsRecordType is the JSONL type for sterile checkpoint // writer diagnostics (queue/barrier/batch/duration aggregates). Emitted by // the CLI after a run when the checkpoint store exposes WriterStats. // Measure-first only: not product throughput marketing. CheckpointWriterStatsRecordType = "gonimbus.reflow.checkpoint_writer_stats.v1" // ObjectPathStageStatsRecordType is the JSONL type for sterile object-path // stage attribution (permit wait, source read, dest write, collision probe, // checkpoint write). Measure-first only: not product throughput marketing. ObjectPathStageStatsRecordType = "gonimbus.reflow.object_path_stage_stats.v1" // WarningRecordType is the JSONL type for transfer reflow warnings. WarningRecordType = "gonimbus.warning.v1" SourceBucketFile = "local" )
const ( // ExecutionPathEngine marks runs executed by the pkg/reflow engine. ExecutionPathEngine = "engine" // ExecutionPathCLIPool marks runs executed by the CLI worker pool. ExecutionPathCLIPool = "cli-pool" )
Execution-path values for the dispatch-transparency contract: every run and summary record names the path that executed it, so path selection between the library engine and the CLI worker pool is observable evidence rather than an implementation detail. Requested concurrency stays in `parallel`; the resolved ceiling and observed max-active live in the embedded ConcurrencyStats.
const ErrCodeInvalidInput = "INVALID_INPUT"
ErrCodeInvalidInput is the error code for a reflow-input record the engine cannot accept. It mirrors the command path's INVALID_INPUT wire value.
const ErrorEventType = "gonimbus.error.v1"
ErrorEventType is the JSONL type for per-object/run reflow error records.
const ReflowInputRecordType = "gonimbus.reflow.input.v1"
ReflowInputRecordType is the JSONL type for a preselected reflow-input record, the line format a RecordStreamSource carries.
Variables ¶
var ErrNotImplemented = errors.New("reflow: runner execution not yet implemented (engine migration in progress)")
ErrNotImplemented is returned by Runner.Run for source forms or scenarios that are still driven by the command layer.
Functions ¶
func ConcurrencyConnectionError ¶
ConcurrencyConnectionError reports provider/transport failures that freeze additive increase without triggering multiplicative decrease.
func DestAdmittedN ¶
func DestAdmittedN(cfg ConcurrencyConfig) int
DestAdmittedN returns the pool/client admitted-N for the destination provider: DestDomainCeilingEffective, floored at 1.
func FormatErrorMessage ¶
FormatErrorMessage combines an operator-facing prefix with a sanitized error.
func IsNonAdvisoryFamily ¶
IsNonAdvisoryFamily reports whether recordType belongs to a recognized work, admission/completeness/integrity, per-object probe, or error record family. Records in these families must never be skipped even if marked advisory=true.
func IsRelativeQuarantinePrefix ¶
IsRelativeQuarantinePrefix reports whether prefix is a relative destination prefix (no leading slash, no URI scheme).
func IsValidPutStorageClass ¶
IsValidPutStorageClass reports whether storageClass is a valid PUT target.
func IsValidRecordTypeToken ¶
IsValidRecordTypeToken reports whether raw is a non-empty, bounded lowercase ASCII record-type identifier token ([a-z0-9._:-], max 128 bytes).
func NewLimiterCopyGate ¶
func NewLimiterCopyGate(limiter *ConcurrencyLimiter, stats *StageStats) transfer.CopyGate
NewLimiterCopyGate returns a transfer.CopyGate that acquires one concurrency token per sequential copy stage and records sterile stage timings.
func NewPathError ¶
func NewPathError(defaultMessage string, verboseMessage string, opts PathErrorOptions) error
NewPathError returns either the default redacted message or the verbose diagnostic message when the caller explicitly opts in.
func ProvenanceNow ¶
func ProvenanceNow() string
ProvenanceNow returns the RFC3339Nano UTC timestamp used for run.ts.
func QuarantineDestRel ¶
QuarantineDestRel joins a relative quarantine prefix with a source key to form the destination-relative key for a quarantined object.
func RequireSourceNewerCapability ¶
RequireSourceNewerCapability validates, from the destination provider's own capability declaration, that it can honor the conditional-write predicates overwrite-if-source-newer needs before any read or destination mutation:
- an If-Match single-PUT compare-and-swap, always; and
- conditional multipart completion when the provider routes large objects through multipart (implements MultipartUploader), so a >threshold overwrite cannot upload parts and then discover the predicate is unsupported only at completion time.
The provider's ConditionalCapabilityReporter declaration is the authority: a provider that does not declare its conditional-write capabilities cannot prove If-Match support and is refused fail-closed. The mere presence of ConditionalPutter is never accepted as proof — an IfAbsent-only implementation exposes ConditionalPutter yet cannot honor If-Match.
A declared predicate must additionally be paired with the callable interface that exercises it. Local method availability is fully knowable and fails closed here, distinct from the remote-endpoint trust boundary (a declaring adapter's remote endpoint is trusted to honor the advertised predicate or return the documented unsupported error). Without this, a contradictory adapter — one that declares If-Match but does not implement ConditionalPutter, or declares conditional multipart completion but does not implement ConditionalMultipartCompleter — would pass the pre-I/O gate and then land a fresh key unconditionally through the head fallback, or upload parts before discovering completion is unavailable. Both the library (validateCollisionCapability) and the command adapter (ensureCollisionCapability) call this so the two surfaces refuse identically.
func SafeReceiptHandle ¶
SafeReceiptHandle preserves an opaque write identifier exactly or omits it. Redacting part of an identifier would falsely identify a different write.
func SafeReceiptSHA256 ¶
SafeReceiptSHA256 accepts only the canonical full-object digest encoding.
func SanitizeOperationCauseMessage ¶
SanitizeOperationCauseMessage returns the redacted provider-error message used in reflow records, warnings, summaries, and checkpoint causes.
func SanitizeRecordType ¶
SanitizeRecordType returns a bounded, printable single-token representation of a record type suitable for logs, warnings, and error messages without echoing arbitrary line/payload content.
func ValidateMetadataBudget ¶
ValidateMetadataBudget enforces the S3 per-pair and total user-metadata byte budget, returning a *MetadataBudgetError when exceeded.
func ValidatePerObjectMetadataRules ¶
func ValidatePerObjectMetadataRules(sourceRules []MetadataSourceKeyRule, derivedRules []MetadataDerivedRule) error
ValidatePerObjectMetadataRules rejects duplicate destination keys across the source-key and derived rule sets.
func ValidateProvenanceModeAndPolicy ¶
ValidateProvenanceModeAndPolicy is the shared validator for the provenance mode and on-write-error policy values. The command adapter delegates to it so the engine and the CLI enforce one rule set.
func ValidateProvenanceSuffix ¶
ValidateProvenanceSuffix enforces the sidecar suffix rules: a leading dot, no path separator, no glob metacharacters, and no collision with a common data extension unless allowUnsafe confirms it.
Types ¶
type AdvisoryWarningTracker ¶
type AdvisoryWarningTracker struct {
// contains filtered or unexported fields
}
AdvisoryWarningTracker bounds per-run warnings for skipped unknown advisory control records to at most one warning per distinct sanitized record type, capped at MaxAdvisoryWarningTypes distinct types per run, while accounting for total skipped records and unreported overflow types.
func NewAdvisoryWarningTracker ¶
func NewAdvisoryWarningTracker() *AdvisoryWarningTracker
NewAdvisoryWarningTracker constructs a per-run tracker for skipped unknown advisory control records.
func (*AdvisoryWarningTracker) CapSummaryWarning ¶
func (t *AdvisoryWarningTracker) CapSummaryWarning() (Warning, bool)
CapSummaryWarning returns an aggregated warning when more than MaxAdvisoryWarningTypes distinct unknown advisory record types were skipped during the run.
func (*AdvisoryWarningTracker) DistinctTypes ¶
func (t *AdvisoryWarningTracker) DistinctTypes() int
DistinctTypes returns the total number of distinct unknown advisory control record types observed during the run.
func (*AdvisoryWarningTracker) Observe ¶
func (t *AdvisoryWarningTracker) Observe(recordType string) (Warning, bool)
Observe records a skipped unknown advisory control record of recordType and returns a sanitized Warning on the first observation of that type (up to MaxAdvisoryWarningTypes distinct types).
func (*AdvisoryWarningTracker) SkippedRecords ¶
func (t *AdvisoryWarningTracker) SkippedRecords() int64
SkippedRecords returns the total number of unknown advisory control records skipped during the run.
func (*AdvisoryWarningTracker) UnreportedRecords ¶
func (t *AdvisoryWarningTracker) UnreportedRecords() int64
UnreportedRecords returns the number of skipped unknown advisory control records whose record type exceeded MaxAdvisoryWarningTypes.
func (*AdvisoryWarningTracker) UnreportedTypes ¶
func (t *AdvisoryWarningTracker) UnreportedTypes() int
UnreportedTypes returns the number of distinct unknown advisory control record types omitted after MaxAdvisoryWarningTypes was reached.
type CheckpointCollision ¶
type CheckpointCollision struct {
DestKey string
Kind string
SourceURI string
SourceETag string
SourceSize int64
DestETag string
DestSize int64
}
CheckpointCollision is the sanitized collision observation an engine hands to a CheckpointStore. It carries only provider-neutral object identity and metadata, never credentials, signed URLs, or raw config.
type CheckpointItem ¶
type CheckpointItem struct {
SourceURI string
DestURI string
SourceKey string
DestKey string
SourceETag string
SourceSize int64
Status string
Reason string
Bytes int64
ErrorCode string
ErrorMessage string
}
CheckpointItem is the sanitized per-item record an engine hands to a CheckpointStore. It carries no credential, signed-URL, or raw-config material.
type CheckpointStore ¶
type CheckpointStore interface {
// ItemDone reports whether a source->dest item completed in a prior run,
// returning the recorded terminal status when done.
ItemDone(ctx context.Context, sourceURI, destURI string) (done bool, status string, err error)
// UpsertItem records the terminal outcome of a reflow item. The item carries
// only sanitized values.
UpsertItem(ctx context.Context, item CheckpointItem) error
// DestKeyObserved reports whether this run/checkpoint has already observed a
// destination key. It backs in-run arbitration when conditional create is not
// the whole decision.
DestKeyObserved(ctx context.Context, destKey string) (bool, error)
// MarkDestKeyObserved records that a destination key has been observed or
// written in this run/checkpoint.
MarkDestKeyObserved(ctx context.Context, destKey string) error
// NoteDestKeySource records the source identity that produced a destination
// key after a successful write.
NoteDestKeySource(ctx context.Context, destKey, sourceURI, sourceETag string, sourceSize int64) error
// NoteCollision records sanitized collision metadata for audit/resume.
NoteCollision(ctx context.Context, collision CheckpointCollision) error
// Close releases store resources.
Close() error
}
CheckpointStore is the minimal resume interface the engine needs. The CLI wraps its storageful (sqlite-backed) implementation behind this interface so the embeddable engine never imports the storage graph (enforced by the dependency-boundary test). Only sanitized values cross this boundary — no credential, signed-URL, or raw-config material. Experimental.
Precondition: the store instance handed to a Runner is already scoped and identity-validated for the compatible run/config by the caller (the CLI adapter performs the resume identity binding before injection). The minimal interface does not itself represent run-identity, so that fail-closed invariant stays with the caller and must not be assumed to be enforced here.
The surface is deliberately narrower than the CLI's sqlite-backed store and grows only with engine-owned resume requirements.
type CheckpointWriterStatsRecord ¶
type CheckpointWriterStatsRecord struct {
MaxBatch int `json:"max_batch"`
QueueDepthSamples int64 `json:"queue_depth_samples"`
QueueDepthSum int64 `json:"queue_depth_sum"`
QueueDepthPeak int64 `json:"queue_depth_peak"`
Admissions int64 `json:"admissions"`
AdmissionWaitNanos int64 `json:"admission_wait_nanos"`
AdmissionWaitMaxNanos int64 `json:"admission_wait_max_nanos"`
AdmissionBlocked int64 `json:"admission_blocked"`
Barriers int64 `json:"barriers"`
BarrierWaitNanos int64 `json:"barrier_wait_nanos"`
BarrierWaitMaxNanos int64 `json:"barrier_wait_max_nanos"`
BarrierOK int64 `json:"barrier_ok"`
BarrierRefusal int64 `json:"barrier_refusal"`
BarrierWriterFailed int64 `json:"barrier_writer_failed"`
BarrierWriterClosed int64 `json:"barrier_writer_closed"`
BarrierCanceled int64 `json:"barrier_canceled"`
Batches int64 `json:"batches"`
BatchSizeSum int64 `json:"batch_size_sum"`
BatchSizeMax int64 `json:"batch_size_max"`
BatchSize1 int64 `json:"batch_size_1"`
BatchSize2To8 int64 `json:"batch_size_2_to_8"`
BatchSize9To32 int64 `json:"batch_size_9_to_32"`
BatchSize33To128 int64 `json:"batch_size_33_to_128"`
BatchSize129Plus int64 `json:"batch_size_129_plus"`
Commits int64 `json:"commits"`
BatchDurationNanos int64 `json:"batch_duration_nanos"`
BatchDurationMaxNanos int64 `json:"batch_duration_max_nanos"`
CommitFatals int64 `json:"commit_fatals"`
RequestRefusals int64 `json:"request_refusals"`
// Experimental savepoint elision counters (Phase A measure).
SavepointsCreated int64 `json:"savepoints_created,omitempty"`
SavepointsElided int64 `json:"savepoints_elided,omitempty"`
}
CheckpointWriterStatsRecord is the payload for gonimbus.reflow.checkpoint_writer_stats.v1. It mirrors pkg/reflowstate WriterStats aggregates without importing that package (dependency boundary). Fields are counts and durations only — no paths, SQL, keys, URIs, or auth.
BatchDuration* is wall time of the full batch write path (BeginTx through per-request SQL/savepoints and tx.Commit), not storage COMMIT alone. Barrier* outcomes are waiter-side observations, not a durable commit ledger. Queue depth samples are approximate.
Experimental: may change with an in-release note.
type CollisionInfo ¶
type CollisionInfo struct {
Kind string `json:"kind"`
DestETagObserved string `json:"dest_etag_observed,omitempty"`
DestSizeObserved *int64 `json:"dest_size_observed,omitempty"`
SrcLastModified *time.Time `json:"src_last_modified,omitempty"`
DestLastModifiedObserved *time.Time `json:"dest_last_modified_observed,omitempty"`
DecisionReason string `json:"decision_reason,omitempty"`
DecisionPath string `json:"decision_path"`
}
CollisionInfo describes the destination collision decision for a reflow item.
type CollisionPolicy ¶
CollisionPolicy selects how the engine handles an existing destination object. Mode is one of skip-if-duplicate|fail|overwrite|quarantine|overwrite-if-source-newer. The engine executes skip-if-duplicate, fail, overwrite, and overwrite-if-source-newer for a supported record-stream destination; quarantine (and other not-yet-migrated cells) return ErrNotImplemented so the caller can fall back to the command path. overwrite-if-source-newer additionally requires a destination that honors the If-Match write precondition (see runRecordStream's capability check).
type ConcurrencyConfig ¶
type ConcurrencyConfig struct {
RequestedCeiling int
EffectiveCeiling int
CeilingReason string
AdaptiveEnabled bool
Floor int
Initial int
MemoryLimitBytes int64
MemoryLimitSource string
MemoryBudgetRequestedBytes int64
MemoryBudgetEffectiveBytes int64
MemoryBudgetSource string
// RetryBufferCapBytes is the resolved per-copy in-memory retry-buffer
// bound (min of the transfer default and the effective budget); the copy
// paths must allocate against this value, spooling above it. Zero means
// not memory-resolved — consumers fall back to the transfer default.
RetryBufferCapBytes int64
// DestDomainMultiplier scales adaptive `current` for the destination
// domain only (GON-067 PR2 dest-biased admission). Zero or one = symmetric
// with source (pre-PR2). Product default is 2 after clamp.
DestDomainMultiplier int
// DestDomainCeilingEffective is the hard dest-domain cap after clamp:
// DestDomainMultiplier * EffectiveCeiling, never silent over-admit.
// Recorded so clients/pools can size to realize the cap.
DestDomainCeilingEffective int
}
ConcurrencyConfig is the resolved GON-048 concurrency contract. RequestedCeiling is the user/operator ceiling; EffectiveCeiling is the resource-safe hard cap.
The Memory* fields record the arithmetic behind a memory-derived ceiling: the detected limit, and the budget (either derived as a fraction of the limit or operator-supplied). Configs constructed directly with an explicit EffectiveCeiling and no probe report zero memory fields — the ceiling was not memory-resolved and the records say so.
func ResolveConcurrency ¶
func ResolveConcurrency(requested int, adaptiveEnabled bool, probe ResourceProbe) ConcurrencyConfig
ResolveConcurrency clamps the requested ceiling by memory and FD limits before the caller allocates concurrency-sized resources. The memory budget is derived as a fraction of the detected limit; use ResolveConcurrencyWithBudget to supply an operator budget instead.
func ResolveConcurrencyWithBudget ¶
func ResolveConcurrencyWithBudget(requested int, adaptiveEnabled bool, probe ResourceProbe, operatorBudgetBytes int64) ConcurrencyConfig
ResolveConcurrencyWithBudget resolves like ResolveConcurrency with an explicit operator memory budget (bytes) replacing the fraction-derived budget. The operator budget is bounded above by the detected memory limit (recorded as operator_clamped_to_limit); when detection is unavailable the operator value is authoritative. operatorBudgetBytes <= 0 means no override. Callers own rejecting invalid operator values before resolution — this function never refuses; it clamps and records.
type ConcurrencyLimiter ¶
type ConcurrencyLimiter struct {
// contains filtered or unexported fields
}
ConcurrencyLimiter gates active copy work and applies AIMD feedback under the resource-derived effective ceiling.
Domain permits (source vs dest) are independent so a sequential Get-then-Put copy path does not couple both endpoints on one global token (GON-067 PR1). Source and probe domains are each capped at adaptive `current`; the dest domain is capped at min(current*DestDomainMultiplier, DestDomainCeilingEffective) (GON-067 PR2 dest-biased residual). Occupancy sums all domains.
func NewConcurrencyLimiter ¶
func NewConcurrencyLimiter(cfg ConcurrencyConfig) *ConcurrencyLimiter
NewConcurrencyLimiter returns an AIMD limiter initialized from cfg. The constructor enforces the same invariant as the runner's config normalization, so a directly-constructed limiter can never observe (or recover to, via throttle) a concurrency above the effective ceiling its snapshots report.
func (*ConcurrencyLimiter) Acquire ¶
func (l *ConcurrencyLimiter) Acquire(ctx context.Context) (func(), error)
Acquire waits until the probe domain permits another active operation (HEAD / collision compare). Copy stages use AcquireSource / AcquireDest so source-read and dest-write are not serialized on one shared token.
func (*ConcurrencyLimiter) AcquireDest ¶
func (l *ConcurrencyLimiter) AcquireDest(ctx context.Context) (func(), error)
AcquireDest admits one destination-side stage (Put / multipart) under the dest-domain cap (adaptive current × DestDomainMultiplier, hard-clamped).
func (*ConcurrencyLimiter) AcquireSource ¶
func (l *ConcurrencyLimiter) AcquireSource(ctx context.Context) (func(), error)
AcquireSource admits one source-side stage (Get / materialize) under the current adaptive ceiling.
func (*ConcurrencyLimiter) DestDomainCeiling ¶
func (l *ConcurrencyLimiter) DestDomainCeiling() int
DestDomainCeiling returns the resolved hard dest-domain admission ceiling (for connection-pool sizing and records).
func (*ConcurrencyLimiter) ObserveConnectionError ¶
func (l *ConcurrencyLimiter) ObserveConnectionError()
ObserveConnectionError freezes additive increase without decreasing current concurrency.
func (*ConcurrencyLimiter) ObserveProviderResult ¶
func (l *ConcurrencyLimiter) ObserveProviderResult(err error)
ObserveProviderResult maps provider outcomes onto the GON-048 signal model.
func (*ConcurrencyLimiter) ObserveSuccess ¶
func (l *ConcurrencyLimiter) ObserveSuccess()
ObserveSuccess records a successful provider operation and may additively increase current concurrency after a clean streak.
func (*ConcurrencyLimiter) ObserveThrottle ¶
func (l *ConcurrencyLimiter) ObserveThrottle()
ObserveThrottle multiplicatively decreases current concurrency.
func (*ConcurrencyLimiter) ReserveCopyMemory ¶
func (l *ConcurrencyLimiter) ReserveCopyMemory(ctx context.Context, sourceSize int64) (func(), error)
ReserveCopyMemory admits a copy's bounded buffer bytes under the effective memory budget before any limiter token or provider action, using the allocator-identical arithmetic (min of known size and the resolved retry cap; unknown sizes reserve the cap). The returned release is exactly-once safe and must run on every terminal path. Configs without a resolved budget return a no-op release.
func (*ConcurrencyLimiter) ResetOccupancyWindow ¶
func (l *ConcurrencyLimiter) ResetOccupancyWindow()
ResetOccupancyWindow restarts the time-averaged occupancy interval. Both execution paths call this immediately before emitting the run record, so the occupancy denominator is the common execution window — setup, preflight, and provider construction are excluded identically on both paths. The run record's startup sample therefore reports zero; the summary value is the completed-run diagnostic.
func (*ConcurrencyLimiter) RetryBufferCap ¶
func (l *ConcurrencyLimiter) RetryBufferCap() int64
RetryBufferCap returns the resolved per-copy in-memory retry-buffer bound the copy paths must allocate against, falling back to the transfer default for configs that were not memory-resolved.
func (*ConcurrencyLimiter) Snapshot ¶
func (l *ConcurrencyLimiter) Snapshot() ConcurrencyStats
Snapshot returns a stable stats view for run and summary records.
type ConcurrencyStats ¶
type ConcurrencyStats struct {
AdaptiveEnabled bool `json:"adaptive_enabled"`
ConcurrencyFloor int `json:"concurrency_floor"`
ConcurrencyInitial int `json:"concurrency_initial"`
ConcurrencyCeilingRequested int `json:"concurrency_ceiling_requested"`
ConcurrencyCeilingEffective int `json:"concurrency_ceiling_effective"`
ConcurrencyCeilingReason string `json:"concurrency_ceiling_reason"`
ConcurrencyFinal int `json:"concurrency_final"`
ConcurrencyThrottleBackoffs int64 `json:"concurrency_throttle_backoffs"`
ConcurrencyAdditiveIncreases int64 `json:"concurrency_additive_increases"`
ConcurrencyConnectionErrorFreezes int64 `json:"concurrency_connection_error_freezes"`
ConcurrencyMaxActive int `json:"concurrency_max_active"`
// ConcurrencyTimeAvgActive is limiter-active provider-operation-seconds
// divided by the common execution window (rounded to 3 decimals) — the
// window starts when the run record is emitted on either path, so setup
// and preflight are excluded identically. The run record's startup sample
// is always 0; the summary value is the completed-run diagnostic. Low
// time-average under a high effective ceiling is a signal to read
// together with throttle/freeze counters and memory-pressure evidence,
// not standalone proof of a starved producer.
ConcurrencyTimeAvgActive float64 `json:"concurrency_time_avg_active"`
// Memory fields are omitted when the ceiling was not memory-resolved
// (configs constructed directly with an explicit EffectiveCeiling and no
// probe); probe-resolved runs always populate limit, budget, and sources.
MemoryLimitBytes int64 `json:"memory_limit_bytes,omitempty"`
MemoryLimitSource string `json:"memory_limit_source,omitempty"`
MemoryBudgetRequestedBytes int64 `json:"memory_budget_requested_bytes,omitempty"`
MemoryBudgetEffectiveBytes int64 `json:"memory_budget_effective_bytes,omitempty"`
MemoryBudgetSource string `json:"memory_budget_source,omitempty"`
RetryBufferCapBytes int64 `json:"retry_buffer_cap_bytes,omitempty"`
// Ledger pressure telemetry: peak outstanding copy-buffer reservation and
// admission-wait evidence. Read together with occupancy — waits with peak
// near the budget indicate memory admission, not a starved producer.
MemoryReservedPeakBytes int64 `json:"memory_reserved_peak_bytes,omitempty"`
MemoryReservationWaits int64 `json:"memory_reservation_waits,omitempty"`
MemoryReservationWaitMS int64 `json:"memory_reservation_wait_ms,omitempty"`
// Dest-domain admission (GON-067 PR2). Multiplier and resolved ceiling are
// static for the run; per-domain peaks prove capacity was used.
DestDomainMultiplier int `json:"dest_domain_multiplier,omitempty"`
DestDomainCeilingEffective int `json:"dest_domain_ceiling_effective,omitempty"`
ConcurrencyMaxActiveSource int `json:"concurrency_max_active_source,omitempty"`
ConcurrencyMaxActiveDest int `json:"concurrency_max_active_dest,omitempty"`
ConcurrencyMaxActiveProbe int `json:"concurrency_max_active_probe,omitempty"`
}
ConcurrencyStats is embedded in reflow run and summary records.
type Config ¶
type Config struct {
// Destination is the structured reflow target.
Destination Destination
// Rewrite controls source->destination key mapping.
Rewrite RewriteConfig
// Collision selects the destination collision policy.
Collision CollisionPolicy
// Concurrency bounds the adaptive copy ceiling. The zero value is resolved to
// defaults by the runner.
Concurrency ConcurrencyConfig
// DryRun plans mappings without writing.
DryRun bool
// ReadOnly blocks provider-side mutations, including write-probe preflight.
ReadOnly bool
// Metadata controls destination metadata and storage-class PUT options.
Metadata MetadataPlan
// Provenance controls per-object provenance sidecar emission. The zero value
// (mode none) disables sidecars.
Provenance ProvenancePlan
// Checkpoint, when non-nil, enables resume via an embedder-supplied store.
Checkpoint CheckpointStore
// Events, when non-nil, receives typed redacted engine events.
Events EventSink
}
Config configures a reflow Runner. It holds injected provider handles and policy; provider construction and credential resolution are caller concerns.
Config implements redacted String/GoString so debug logging cannot dump an injected provider or credential handle.
This surface carries the structural knobs that have migrated into the engine. Provenance and file-source policy fields are added alongside their decision logic rather than designed speculatively here. Experimental.
type DestLayout ¶
type DestLayout struct {
ProviderID string
BaseURI string
// Object-store decomposition.
Bucket string
Prefix string
// File decomposition.
BaseDir string
}
DestLayout is the resolved destination key/URI layout for a reflow run: the provider id plus the structural decomposition of the base URI (bucket+prefix for object stores, base directory for file). It produces destination keys and URIs deterministically — no provider I/O. Construct it with ParseDestLayout from the structured Destination's BaseURI. Experimental.
func ParseDestLayout ¶
func ParseDestLayout(baseURI string) (DestLayout, error)
ParseDestLayout decomposes a destination base URI into the layout the engine keys writes against. It mirrors the CLI's destination parsing: a file URI yields a cleaned base directory; an object-store URI yields bucket+prefix with a trailing slash. It performs no I/O.
func (DestLayout) DestKey ¶
func (l DestLayout) DestKey(destRel string) string
DestKey maps a destination-relative key to the full destination object key.
func (DestLayout) DestURI ¶
func (l DestLayout) DestURI(destKey string) string
DestURI renders the full destination URI for a destination object key.
type Destination ¶
type Destination struct {
// Provider is the injected destination provider handle.
Provider provider.Provider
// ProviderID is the internal provider id ("s3", "file", "gcs").
ProviderID string
// BaseURI is the destination base prefix, e.g. s3://bucket/base/ or
// file:///tmp/out/.
BaseURI string
}
Destination is the structured reflow target: an injected provider handle, the provider id (e.g. "s3", "file", "gcs"), and the base URI/prefix that rewritten keys are written under. Provider construction details (region, profile, endpoint, credentials) are a caller concern resolved before the handle is injected — embedders do not reconstruct command flags. Experimental.
func (Destination) GoString ¶
func (d Destination) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (Destination) String ¶
func (d Destination) String() string
String returns a redacted summary that never exposes the injected provider handle, which may hold credential material (TokenSource/CredentialsProvider). Mirrors the redaction discipline of the pkg/provider config types.
type EnvelopeInspection ¶
type EnvelopeInspection struct {
Type string
SanitizedType string
Data json.RawMessage
Advisory bool
AdvisoryMalformed bool
DuplicateEnvelopeKey string
ProducerVersion string
UnmarshalErr error
}
EnvelopeInspection holds the parsed top-level JSONL record envelope fields used by reflow input classifiers and parsers across all three intake boundaries (classifyReflowFirstRecord, CLI enqueueReflowLine, and engine parseReflowInputLine).
func InspectRecordEnvelope ¶
func InspectRecordEnvelope(line string) EnvelopeInspection
InspectRecordEnvelope parses the top-level JSONL envelope of line, validating that any present top-level "advisory" field is a strict JSON boolean and rejecting case-variant or duplicate top-level envelope keys.
func (EnvelopeInspection) FormatAdvisoryMalformedError ¶
func (e EnvelopeInspection) FormatAdvisoryMalformedError() error
FormatAdvisoryMalformedError returns an attributable error when the top-level advisory field is present but not a JSON boolean or when duplicate top-level envelope keys are present.
func (EnvelopeInspection) FormatCLIUnknownTypeError ¶
func (e EnvelopeInspection) FormatCLIUnknownTypeError() error
FormatCLIUnknownTypeError returns the CLI pool's attributable fail-closed error for an unsupported JSON record type.
func (EnvelopeInspection) FormatEngineUnknownTypeError ¶
func (e EnvelopeInspection) FormatEngineUnknownTypeError() error
FormatEngineUnknownTypeError returns the engine's attributable fail-closed error for an unsupported record type.
func (EnvelopeInspection) FormatFirstRecordUnknownTypeReason ¶
func (e EnvelopeInspection) FormatFirstRecordUnknownTypeReason(malformedPrefix, unknownTypePrefix string) string
FormatFirstRecordUnknownTypeReason returns the pre-execution first-record sniff refusal reason for an unsupported JSON record type.
func (EnvelopeInspection) IsSkippableAdvisoryUnknown ¶
func (e EnvelopeInspection) IsSkippableAdvisoryUnknown() bool
IsSkippableAdvisoryUnknown reports whether an unknown record envelope (after known exact record types have been dispatched first) may be skipped under the advisory control-record contract.
type ErrorEvent ¶
type ErrorEvent struct {
Code string `json:"code"`
Key string `json:"key,omitempty"`
Message string `json:"message"`
Details map[string]any `json:"details,omitempty"`
Collision any `json:"collision,omitempty"`
}
ErrorEvent is the redacted payload describing a per-object or run error. Message and Details are sanitized (provider-error redaction applied) by the engine before delivery, so a sink may log or persist them without further scrubbing.
type EventSink ¶
type EventSink interface {
// OnRun reports the resolved run configuration.
OnRun(ctx context.Context, rec RunRecord) error
// OnSource reports the resolved source.
OnSource(ctx context.Context, rec SourceRunRecord) error
// OnRecord reports a per-object outcome.
OnRecord(ctx context.Context, rec Record) error
// OnWarning reports a non-fatal condition (e.g. the IfAbsent fallback notice).
OnWarning(ctx context.Context, w Warning) error
// OnError reports a per-object or run error with a redacted message.
OnError(ctx context.Context, e ErrorEvent) error
// OnSummary reports the terminal run summary.
OnSummary(ctx context.Context, rec SummaryRecord) error
}
EventSink receives typed engine events. The engine applies its redaction boundary before the sink — every payload is already sanitized, so a sink may log or persist it without re-scrubbing.
EventSink is the engine's output adapter boundary (the CLI implements it over a JSONL writer), so each method returns an error: a sink that fails to write or persist an event surfaces that failure to the engine, which can abort and classify it the way the terminal output path does. A nil EventSink disables event delivery. Experimental.
type FileTreeSource ¶
type FileTreeSource struct {
Root string
}
FileTreeSource reflows a local filesystem tree rooted at Root.
func (FileTreeSource) GoString ¶
func (s FileTreeSource) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (FileTreeSource) String ¶
func (s FileTreeSource) String() string
String renders the root by presence only: a local filesystem path is treated as disclosure-sensitive, consistent with the engine's local-path discretion.
type IfAbsentCapability ¶
type IfAbsentCapability struct {
ProbeStatus preflight.IfAbsentProbeStatus
Honored *bool
FallbackActive bool
ProbeError string
}
IfAbsentCapability records what the engine knows about a destination's conditional-create (If-None-Match: *) support for the run, and whether the head-compare fallback is active. Experimental.
type InvalidInputsError ¶
type InvalidInputsError struct{ Count int64 }
InvalidInputsError reports that a run completed and emitted its terminal summary but encountered one or more invalid input records (surfaced as INVALID_INPUT events). The Summary is still returned with it. It mirrors the command path, which exits non-zero when invalid inputs occur, so a library caller does not observe success for a stream the CLI reports as failed.
func (*InvalidInputsError) Error ¶
func (e *InvalidInputsError) Error() string
type LaneCheckpointStore ¶
type LaneCheckpointStore interface {
CheckpointStore
producer.DurableStore
// UnitDone reconciles a previously committed terminal by stable unit key.
UnitDone(ctx context.Context, unitKey string) (done bool, status string, err error)
// ResumeEnabled reports whether this scoped store is being used for replay.
ResumeEnabled() bool
// CheckpointUnit atomically records the terminal item and acknowledges the
// unit. It returns only after the shared durable transaction commits.
CheckpointUnit(ctx context.Context, item CheckpointItem, unitKey string) error
}
LaneCheckpointStore is the optional keyed durability capability required by live partitioned sources. It combines producer state with the downstream checkpoint so one transaction can record an item terminal and acknowledge its stable work-unit key. Experimental.
type MetadataBudgetError ¶
type MetadataBudgetError struct {
OverLimitKeys []string
PairLimit int
TotalBytes int
TotalLimit int
Count int
}
MetadataBudgetError reports user metadata exceeding the S3 metadata budget.
func (*MetadataBudgetError) Details ¶
func (e *MetadataBudgetError) Details() map[string]any
Details renders the structured detail map for error reporting.
func (*MetadataBudgetError) Error ¶
func (e *MetadataBudgetError) Error() string
type MetadataDerivationError ¶
type MetadataDerivationError struct {
DestKey string
Expression string
Reason string
ResultKind string
CollidingKeys []string
}
MetadataDerivationError describes a per-object metadata derivation failure under --metadata-on-missing-source=fail.
func (*MetadataDerivationError) Details ¶
func (e *MetadataDerivationError) Details() map[string]any
Details renders the structured detail map for error reporting.
func (*MetadataDerivationError) Error ¶
func (e *MetadataDerivationError) Error() string
type MetadataDerivedRule ¶
type MetadataDerivedRule struct {
DestKey string
Expression string
Raw string
// contains filtered or unexported fields
}
MetadataDerivedRule projects a compiled expression over a source object onto a destination user-metadata key. Construct it via ParseMetadataDerivedRules: the compiled expression is held in an unexported field, so a hand-built rule has no expression and is rejected by ValidatePerObjectMetadataRules (and fails closed in PutOptions) rather than panicking.
func ParseMetadataDerivedRules ¶
func ParseMetadataDerivedRules(raw []string) ([]MetadataDerivedRule, error)
ParseMetadataDerivedRules compiles dest=expression projection entries.
type MetadataPlan ¶
type MetadataPlan struct {
Policy string
Set map[string]string
SourceKeyRules []MetadataSourceKeyRule
DerivedRules []MetadataDerivedRule
OnMissingSource string
PreserveContentType bool
DestinationStorageClass string
MetadataSidecarSuffix string
}
MetadataPlan is the resolved, flag-free destination-metadata decision plan. It carries the policy plus the compiled per-object rules and produces the provider.PutOptions for a source object deterministically — no provider I/O. The CLI resolves it from flags; the engine consumes it. Experimental.
func (MetadataPlan) GoString ¶
func (c MetadataPlan) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (MetadataPlan) HasPerObjectRules ¶
func (c MetadataPlan) HasPerObjectRules() bool
HasPerObjectRules reports whether any per-object metadata rule is configured.
func (MetadataPlan) NeedsSourceHead ¶
func (c MetadataPlan) NeedsSourceHead() bool
NeedsSourceHead reports whether resolving the plan for an object requires a source HEAD (to read source metadata/content-type/storage-class).
func (MetadataPlan) PutOptions ¶
func (c MetadataPlan) PutOptions(source *provider.ObjectMeta) (provider.PutOptions, error)
PutOptions builds the destination provider.PutOptions for a source object, applying the policy, per-object rules, content-type, and storage-class decisions deterministically. source may be nil only for policies that do not read source metadata.
func (MetadataPlan) RequiresCapability ¶
func (c MetadataPlan) RequiresCapability() bool
RequiresCapability reports whether the plan requires a metadata-aware PUT.
func (MetadataPlan) String ¶
func (c MetadataPlan) String() string
String returns a value-free summary suitable for debug output. Metadata values and derived expression text may reflect operator-sensitive configuration, so they are never formatted by value.
func (MetadataPlan) Validate ¶
func (c MetadataPlan) Validate() error
Validate checks the resolved plan's enum fields and per-object rules. It is the canonical resolved-plan validation: the CLI flag-resolution path delegates to it, and direct library callers should call it before handing the plan to the engine. PutOptions additionally fails closed on the policy and fixed storage-class fields as defense-in-depth, but Validate is the single place that rejects an inconsistent plan up front. It does not validate the file-sidecar suffix, which is a destination-provider concern carried by the CLI.
type MetadataRunConfig ¶
type MetadataRunConfig struct {
Policy string `json:"policy"`
SetKeys []string `json:"set_keys,omitempty"`
SourceKeyRuleKeys []string `json:"source_key_rule_keys,omitempty"`
DerivedRuleKeys []string `json:"derived_rule_keys,omitempty"`
OnMissingSource string `json:"on_missing_source,omitempty"`
PreserveContentType bool `json:"preserve_content_type,omitempty"`
DestinationStorageClass string `json:"destination_storage_class,omitempty"`
MetadataSidecarSuffix string `json:"metadata_sidecar_suffix,omitempty"`
Set map[string]string `json:"set,omitempty"`
}
MetadataRunConfig describes destination metadata behavior for a reflow run.
type MetadataSourceKeyRule ¶
MetadataSourceKeyRule projects a single source user-metadata key onto a destination user-metadata key.
func ParseMetadataSourceKeyRules ¶
func ParseMetadataSourceKeyRules(raw []string) ([]MetadataSourceKeyRule, error)
ParseMetadataSourceKeyRules compiles dest=source projection entries.
type MissingConditionalCapabilityError ¶
type MissingConditionalCapabilityError struct {
// Mode is the collision mode that required the missing predicate.
Mode string
// MissingCapability is the provider-dispatch capability label, e.g.
// "ConditionalPutter.IfMatchETag" or "ConditionalMultipartCompleter.IfMatchETag".
MissingCapability string
// contains filtered or unexported fields
}
MissingConditionalCapabilityError reports that a destination provider cannot honor a conditional-write predicate a collision mode requires. It carries the provider-dispatch capability label so the command adapter can surface a structured missing_capability preflight detail identical to its other capability errors, and both the library and the command adapter derive it from the same RequireSourceNewerCapability authority.
func (*MissingConditionalCapabilityError) Error ¶
func (e *MissingConditionalCapabilityError) Error() string
type ObjectErrorsError ¶
type ObjectErrorsError struct{ Count int64 }
ObjectErrorsError reports that a run completed and emitted its terminal summary but encountered one or more object-level errors. The Summary is still returned with it. It mirrors the command path, which exits non-zero when per-object errors occur, so a library caller does not observe success for a run the CLI reports as failed.
func (*ObjectErrorsError) Error ¶
func (e *ObjectErrorsError) Error() string
type ObjectPathStageStatsEmitter ¶
type ObjectPathStageStatsEmitter interface {
OnObjectPathStageStats(ctx context.Context, rec ObjectPathStageStatsRecord) error
}
ObjectPathStageStatsEmitter is an optional EventSink extension. When the engine's Events value implements it, finishRun emits sterile stage stats after the summary. CLI JSONL adapters implement this; test sinks need not.
type ObjectPathStageStatsRecord ¶
type ObjectPathStageStatsRecord struct {
PermitWaits int64 `json:"permit_waits"`
PermitWaitNanos int64 `json:"permit_wait_nanos"`
PermitWaitMaxNanos int64 `json:"permit_wait_max_nanos"`
SourceReads int64 `json:"source_reads"`
SourceReadNanos int64 `json:"source_read_nanos"`
SourceReadMaxNanos int64 `json:"source_read_max_nanos"`
DestWrites int64 `json:"dest_writes"`
DestWriteNanos int64 `json:"dest_write_nanos"`
DestWriteMaxNanos int64 `json:"dest_write_max_nanos"`
CoupledCopies int64 `json:"coupled_copies"`
CoupledCopyNanos int64 `json:"coupled_copy_nanos"`
CoupledCopyMaxNanos int64 `json:"coupled_copy_max_nanos"`
CollisionProbes int64 `json:"collision_probes"`
CollisionProbeNanos int64 `json:"collision_probe_nanos"`
CollisionProbeMaxNanos int64 `json:"collision_probe_max_nanos"`
CheckpointWrites int64 `json:"checkpoint_writes"`
CheckpointWriteNanos int64 `json:"checkpoint_write_nanos"`
CheckpointWriteMaxNanos int64 `json:"checkpoint_write_max_nanos"`
// PhaseSplitCopies counts copies that used independent source-read and
// dest-write permits (the default single-part path after PR1).
PhaseSplitCopies int64 `json:"phase_split_copies"`
// CoupledCopyOps counts copies that held one permit across source+dest
// (streaming multipart / coupled path).
CoupledCopyOps int64 `json:"coupled_copy_ops"`
// Per-domain permit waits (PR2): prove dest capacity was admitted separately
// from source. Aggregate PermitWaits remains the sum of domains.
SourcePermitWaits int64 `json:"source_permit_waits"`
SourcePermitWaitNanos int64 `json:"source_permit_wait_nanos"`
SourcePermitWaitMaxNanos int64 `json:"source_permit_wait_max_nanos"`
DestPermitWaits int64 `json:"dest_permit_waits"`
DestPermitWaitNanos int64 `json:"dest_permit_wait_nanos"`
DestPermitWaitMaxNanos int64 `json:"dest_permit_wait_max_nanos"`
ProbePermitWaits int64 `json:"probe_permit_waits"`
ProbePermitWaitNanos int64 `json:"probe_permit_wait_nanos"`
ProbePermitWaitMaxNanos int64 `json:"probe_permit_wait_max_nanos"`
// Peak concurrent tokens observed per domain (from limiter snapshot).
SourceMaxActive int `json:"source_max_active"`
DestMaxActive int `json:"dest_max_active"`
ProbeMaxActive int `json:"probe_max_active"`
// Resolved dest-domain policy for the run (mirrors ConcurrencyStats).
DestDomainMultiplier int `json:"dest_domain_multiplier,omitempty"`
DestDomainCeilingEffective int `json:"dest_domain_ceiling_effective,omitempty"`
}
ObjectPathStageStatsRecord is the payload for gonimbus.reflow.object_path_stage_stats.v1. Counts and durations only — no paths, keys, URIs, SQL, or auth. Emitted after the summary when stage attribution was collected for the run.
Experimental: may change with an in-release note.
type ObjectSource ¶
ObjectSource reflows a single object addressed by URI from an injected provider.
URI must be the object AS SPELLED, not a canonicalized form. Parsing is escape-aware: an escaped metacharacter (`file\*.txt`) is a literal object-key character, and a caller that canonicalizes the URI first would strip the escape, leaving a spelling the engine reclassifies as a pattern and refuses.
func (ObjectSource) GoString ¶
func (s ObjectSource) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (ObjectSource) String ¶
func (s ObjectSource) String() string
String returns a redacted summary that never exposes the injected provider handle, which may hold credential material.
type PathErrorOptions ¶
type PathErrorOptions struct {
Verbose bool
}
PathErrorOptions controls how local path errors cross the library boundary.
type PrefixSource ¶
type PrefixSource struct {
Provider provider.Provider
URI string
Enumerator producer.LaneEnumerator
Authority *partition.Authority
}
PrefixSource reflows every object under a listing prefix, optionally narrowed by a glob pattern, from an injected provider.
URI must be the selector AS SPELLED, for the reason ObjectSource states: the parse is escape-aware, and a canonicalized spelling has already lost the escapes that decide whether a metacharacter is a glob or a literal key character. An exact-object URI is refused here — ObjectSource is that form — because listing under a whole object key would also admit its prefix siblings.
Provider is required for dry-run as well as copy, unlike ObjectSource. A prefix names no object until it is listed, so List is the planning operation itself and there is no plan to produce without it.
Enumeration is page-streamed: each listed object is dispatched as it is read, so a selector matching many objects does not accumulate a listing. A List failure partway through stops enumeration and fails the run with a SourceEnumerationError; see that type for what the run does and does not report in that case.
Enumerator and Authority are optional only as a pair. When both are nil the standing serial provider-page path remains active. When both are set, Enumerator supplies partitioned objects and the Runner independently checks every carried LaneRef against Authority before worker dispatch. The caller, not the enumerator, owns that authority so an identity claim cannot authorize itself.
func (PrefixSource) GoString ¶
func (s PrefixSource) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (PrefixSource) String ¶
func (s PrefixSource) String() string
String returns a redacted summary that never exposes the injected provider handle, which may hold credential material.
type ProvenanceEmitter ¶
type ProvenanceEmitter interface {
// EmitProvenanceWarning reports a warn-policy sidecar write failure.
EmitProvenanceWarning(ctx context.Context, w Warning) error
// EmitProvenanceError reports a fail-policy sidecar write failure.
EmitProvenanceError(ctx context.Context, key, message string, cause error, details map[string]any) error
}
ProvenanceEmitter delivers the sidecar writer's warn/fail outcomes through the calling surface's native record path, so each surface keeps its own record shape (the command pool emits its error record; the engine emits an ErrorEvent through its EventSink) while sharing the content and policy.
type ProvenancePlacementContext ¶
type ProvenancePlacementContext struct {
Mode string `json:"mode"`
SidecarRoot string `json:"sidecar_root,omitempty"`
}
ProvenancePlacementContext describes where provenance sidecars are written.
type ProvenancePlacementPlan ¶
type ProvenancePlacementPlan struct {
Mode string
SidecarRoot *ProvenanceSidecarRoot
// SidecarProvider is an injected second provider handle used only for a
// file mirrored-root, whose sidecars land under a base directory distinct
// from the destination provider's. Sibling and same-bucket object-store
// mirrored placement write through the destination handle, so this stays
// nil for them. It is redaction-safe: String/GoString expose presence only.
SidecarProvider provider.Provider
}
ProvenancePlacementPlan is the resolved sidecar placement. SidecarRoot is set only for mirrored-root placement.
type ProvenancePlan ¶
type ProvenancePlan struct {
Mode string
Suffix string
AllowUnsafeSuffix bool
OnWriteError string
Placement ProvenancePlacementPlan
RunID string
ToolVersion string
}
ProvenancePlan is the resolved, validated provenance-sidecar execution plan and the library-owned authority for a run. Validate is canonical: the command adapter delegates to it, and the runner calls it before any stream read, event emission, provider probe, or destination mutation. A plan carries explicit run identity/version (audit anchors, never reached from process globals) and the explicit unsafe-suffix confirmation, so an accidental data-extension suffix cannot pass as an approved one.
func (ProvenancePlan) Enabled ¶
func (p ProvenancePlan) Enabled() bool
Enabled reports whether the plan writes sidecars.
func (ProvenancePlan) GoString ¶
func (p ProvenancePlan) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (ProvenancePlan) RunConfig ¶
func (p ProvenancePlan) RunConfig() *ProvenanceRunConfig
RunConfig derives the gonimbus.reflow.run.v1 provenance echo from the validated plan. The engine emits this in its RunRecord so a direct library caller and the command share one authority (the command no longer overwrites it). Returns nil when provenance is disabled.
func (ProvenancePlan) String ¶
func (p ProvenancePlan) String() string
String renders a redacted summary: the injected sidecar provider handle is never dumped (presence only), while the non-secret run identity/version and placement remain visible for debugging.
func (ProvenancePlan) Validate ¶
func (p ProvenancePlan) Validate() error
Validate enforces the provenance plan's internal consistency before any I/O.
func (ProvenancePlan) ValidateAgainstDestination ¶
func (p ProvenancePlan) ValidateAgainstDestination(destProviderID, destBucket string) error
ValidateAgainstDestination verifies an object-store mirrored sidecar root against the run's actual resolved destination (provider id + bucket), so a caller's SameBucketAsDest assertion is proven rather than trusted. It runs at the runner boundary, before any I/O: a self-asserted same-bucket root that in fact names a different provider or bucket is refused, closing the fail-closed authority- integrity gap that ProvenancePlan.Validate (which has no destination) cannot see. The file mirrored root has no bucket; its authority is the injected provider, checked by the callable-ObjectPutter admission.
type ProvenancePutFunc ¶
ProvenancePutFunc writes a marshaled sidecar payload to the sidecar key. It is the injected write transport so the shared writer stays transport-agnostic: the engine passes a function that routes the PUT through its concurrency/AIMD limiter (Acquire + ObserveProviderResult); the command pool passes the provider's PutObject directly (ProvenancePutViaProvider). A nil ProvenancePutFunc means the sidecar destination cannot accept a PUT and is reported through the on-write-error policy.
func ProvenancePutViaProvider ¶
func ProvenancePutViaProvider(p provider.Provider) ProvenancePutFunc
ProvenancePutViaProvider adapts a provider's ObjectPutter to a ProvenancePutFunc, returning nil when the provider does not support PutObject so the shared writer reports the unsupported-destination failure through the on-write-error policy.
type ProvenanceRef ¶
type ProvenanceRef struct {
Written bool `json:"written"`
Key string `json:"key"`
URI string `json:"uri,omitempty"`
}
ProvenanceRef identifies a provenance sidecar emitted for a reflow item.
func WriteProvenanceSidecar ¶
func WriteProvenanceSidecar(ctx context.Context, put ProvenancePutFunc, runID, toolVersion, ts, mode, onWriteError, sidecarKey, sidecarURI, destURI string, in ProvenanceSidecarInput, emit ProvenanceEmitter) (ref *ProvenanceRef, sidecarFatal bool, emitErr error)
WriteProvenanceSidecar builds the audit payload, writes it through put, and applies the on-write-error policy. It returns the reference (with Written set), whether the failure is fatal to the item (fail policy), and any event-sink delivery error. The sidecarKey/sidecarURI are resolved by the caller from its placement layout; mode labels the operational surface in the emitted details. Emission goes through emit so each surface keeps its native record shape.
The event-sink delivery error is propagated (never swallowed): a warn- or fail-policy write failure whose warning/error event could not be delivered must not let a caller silently proceed to a success terminal, so the caller treats a non-nil emitErr as fatal to the item (fatal-to-item ordering).
type ProvenanceRunConfig ¶
type ProvenanceRunConfig struct {
Mode string `json:"mode"`
Suffix string `json:"suffix,omitempty"`
OnWriteError string `json:"on_write_error,omitempty"`
Placement ProvenancePlacementContext `json:"placement"`
}
ProvenanceRunConfig describes provenance behavior for a reflow run.
type ProvenanceSidecarInput ¶
type ProvenanceSidecarInput struct {
SourceURI string
SourceETag string
SourceSize int64
SourceLastMod time.Time
DestURI string
DestETag string
DestSize int64
RoutingClass string
// RewriteTemplate is the rewrite "to" template, when one was used.
RewriteTemplate string
// QuarantinePrefix is set only for quarantine routing.
QuarantinePrefix string
Collision *CollisionInfo
Vars map[string]string
Probe *probe.ProbeAudit
// Action is the pool action string recorded in the sidecar (e.g. the landed
// action, "skipped.duplicate", or "quarantined").
Action string
}
ProvenanceSidecarInput carries the per-object facts a provenance sidecar records, expressed independently of the CLI-pool or engine input types so a single builder serves both surfaces. The command adapter fills it from its reflowTask; the engine fills it from a reflowInput.
type ProvenanceSidecarPayload ¶
type ProvenanceSidecarPayload struct {
Schema string `json:"schema"`
SchemaVersion string `json:"schema_version"`
Source provenanceSourceBlock `json:"source"`
Destination provenanceDestBlock `json:"destination"`
Run provenanceRunBlock `json:"run"`
Routing provenanceRoutingBlock `json:"routing"`
Collision *CollisionInfo `json:"collision,omitempty"`
Vars map[string]string `json:"vars,omitempty"`
Probe *probe.ProbeAudit `json:"probe,omitempty"`
Action string `json:"action"`
}
ProvenanceSidecarPayload is the gonimbus.provenance.v1 audit object. Its JSON serialization is the wire contract compared byte-for-byte across execution paths (modulo the intentionally time-varying run.ts).
func BuildProvenanceSidecar ¶
func BuildProvenanceSidecar(runID, toolVersion, ts string, in ProvenanceSidecarInput) ProvenanceSidecarPayload
BuildProvenanceSidecar assembles the audit payload from neutral inputs. runID and toolVersion are supplied by the caller (the run identity/version are explicit inputs, never reached from process globals). ts is the run timestamp; callers pass a stable value so parity comparisons can normalize only this field.
type ProvenanceSidecarRoot ¶
type ProvenanceSidecarRoot struct {
Provider string
Bucket string
Prefix string
BaseDir string
BaseURI string
SameBucketAsDest bool
}
ProvenanceSidecarRoot is a resolved mirrored sidecar-root layout. For object stores it is constrained to the destination bucket (SameBucketAsDest true); a cross-bucket object-store root is refused at resolution, so a sidecar URI is never rendered from a bucket other than the one written through.
type Record ¶
type Record struct {
SourceURI string `json:"source_uri"`
SourceBucket string `json:"source_bucket,omitempty"`
SourceRoot string `json:"source_root,omitempty"`
SourceKey string `json:"source_key"`
SourceETag string `json:"source_etag,omitempty"`
SourceSize int64 `json:"source_size_bytes,omitempty"`
SourceLastModified string `json:"source_last_modified,omitempty"`
DestSHA256 string `json:"dest_sha256,omitempty"`
DestVersionID string `json:"dest_version_id,omitempty"`
DestETag string `json:"dest_etag,omitempty"`
DestURI string `json:"dest_uri"`
DestKey string `json:"dest_key"`
Bytes int64 `json:"bytes,omitempty"`
Status string `json:"status"`
Reason string `json:"reason,omitempty"`
RoutingClass string `json:"routing_class,omitempty"`
Collision *CollisionInfo `json:"collision,omitempty"`
Provenance *ProvenanceRef `json:"provenance,omitempty"`
Details map[string]any `json:"details,omitempty"`
}
Record is the payload for gonimbus.reflow.v1 JSONL records.
func RecordWithReceipt ¶
func RecordWithReceipt(rec Record, receipt transfer.CopyReceipt, destProvider string) Record
RecordWithReceipt projects only successful write facts onto a terminal. Source read metadata takes precedence over an existing input observation.
func (Record) MarshalJSON ¶
type RecordStreamSource ¶
type RecordStreamSource struct {
Records io.Reader
Resolve SourceResolver
}
RecordStreamSource reflows a preselected stream of reflow-input records — the library equivalent of `crawl --emit reflow-input | transfer reflow --stdin`.
The engine consumes the stream record-by-record (it never materializes the whole stream). The current implementation executes the dry-run and copy planes for S3 gonimbus.reflow.input.v1 records. Dry-run does not invoke Resolve; copy execution requires it so a source provider is obtained per record. A record outside the supported subset surfaces as an INVALID_INPUT event rather than being planned.
func (RecordStreamSource) GoString ¶
func (s RecordStreamSource) GoString() string
GoString implements fmt %#v with the same redaction as String.
func (RecordStreamSource) String ¶
func (s RecordStreamSource) String() string
String renders presence only: the record stream and resolver may close over credential material, so neither is formatted by value.
type ResourceProbe ¶
type ResourceProbe struct {
MemoryLimitBytes func() (int64, string, error)
FDSoftLimit func() (int64, error)
}
ResourceProbe supplies host resource limits used to clamp requested transfer concurrency before any worker pools, queues, or retry buffers are created.
func DefaultResourceProbe ¶
func DefaultResourceProbe() ResourceProbe
DefaultResourceProbe returns the platform probes used by transfer reflow.
type RewriteConfig ¶
RewriteConfig holds the segment-capture rewrite templates (the library form of --rewrite-from / --rewrite-to). An empty pair preserves source-relative keys.
type RunRecord ¶
type RunRecord struct {
DestURI string `json:"dest_uri"`
CheckpointPath string `json:"checkpoint_path"`
DryRun bool `json:"dry_run"`
Resume bool `json:"resume"`
Parallel int `json:"parallel"`
ExecutionPath string `json:"execution_path"`
ConcurrencyStats
Provenance *ProvenanceRunConfig `json:"provenance,omitempty"`
Metadata *MetadataRunConfig `json:"metadata,omitempty"`
}
RunRecord is the payload for gonimbus.reflow.run.v1 JSONL records.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner executes reflow runs against an injected destination/provider matrix. Construct one with NewRunner. Experimental.
func (*Runner) Run ¶
Run executes the reflow for src and returns a Summary. When Config.Events is set, the typed event stream carries the run/source records, per-object Records, Warnings, Errors, and the terminal Summary — all redacted before delivery.
Experimental: the engine migration is landing incrementally. Run executes the migrated paths (currently dry-run and copy over a RecordStreamSource) and returns ErrNotImplemented for forms and scenarios still driven by the command layer, so callers can fall back to the CLI path for those.
type Source ¶
type Source interface {
// contains filtered or unexported methods
}
Source identifies what a Runner reads. It is a closed set of typed forms — ObjectSource, PrefixSource, FileTreeSource, and RecordStreamSource — so an embedder selects a form rather than reconstructing command flags. Experimental.
type SourceEnumerationError ¶
SourceEnumerationError reports that listing a source selector failed partway through: the engine stopped requesting pages, the work it had already admitted drained to its own terminal records and checkpoint dispositions, and the run then failed WITHOUT emitting a terminal summary.
The withheld summary is the point. A summary accounts for a whole selector, and a run that stopped enumerating mid-selector has no such accounting to offer; emitting one invites a consumer to read a partial count as a complete one. This is a deliberate divergence from the CLI pool, which writes its summary after the same drain.
It is a run-level failure, not an object terminal: no per-object error event is synthesized for the objects that were never listed. Selector names what was being enumerated (sanitized), and Unwrap exposes the provider's own error, so a caller classifies the cause with errors.Is/errors.As rather than by matching message text.
func (*SourceEnumerationError) Error ¶
func (e *SourceEnumerationError) Error() string
func (*SourceEnumerationError) Unwrap ¶
func (e *SourceEnumerationError) Unwrap() error
type SourceMetadataCollisionError ¶
type SourceMetadataCollisionError struct {
Keys []string
}
SourceMetadataCollisionError reports source metadata keys that collide after canonicalization (lower-cased, trimmed).
func (*SourceMetadataCollisionError) Error ¶
func (e *SourceMetadataCollisionError) Error() string
type SourceResolver ¶
SourceResolver maps a parsed source URI to a source provider handle for RecordStreamSource inputs. Returning an error fails (or skips, per source policy) the affected record.
Concurrency contract: the engine invokes the resolver only from its serial planning stage, never concurrently — implementations (including lazy provider caches) need no internal synchronization. Record execution fans out to a worker pool, but each record's resolved handle is captured before dispatch, so the returned provider must tolerate concurrent object operations (the standing provider contract) while the resolver itself does not need to.
type SourceRunMetadata ¶
SourceRunMetadata is the resolved run-level description of the source a positional run selected, for persistence in a trusted checkpoint store. It carries no credential, signed-URL, or raw-config material — but it is NOT an event or log payload: Root and the selector can be disclosure-sensitive (a local filesystem root on the file path) and a store records them verbatim.
URI is the run-level source SELECTOR — for a prefix or pattern source it is the prefix/pattern itself, which matches many objects and matches no single one. It is never item resume authority: ItemDone, UpsertItem, collision, and destination-source entries are keyed by each enumerated object's exact object URI. Reconstructing one identity from the other would either make resume lookups miss or make the run-level source record misdescribe what was selected.
type SourceRunMetadataStore ¶
type SourceRunMetadataStore interface {
// SetSourceRunMetadata records the resolved run-level source.
SetSourceRunMetadata(ctx context.Context, meta SourceRunMetadata) error
}
SourceRunMetadataStore is an optional CheckpointStore capability: a store that records the run-level source a positional run resolved. Positional execution type-asserts for it when wired and skips the write for a store that does not implement it — absence is a clean no-op, not a deficiency, because a record-stream run has no single positional selector to record. Experimental.
This is run-level SELECTOR metadata. It is unrelated to the per-object source metadata of metadata.go (user metadata copied or derived into destination PUT options); the two share a word, not a contract.
The write is run setup performed through the coordinator, ordered ahead of any per-item authority write.
type SourceRunRecord ¶
type SourceRunRecord struct {
Provider string `json:"provider"`
Bucket string `json:"source_bucket,omitempty"`
Root string `json:"source_root,omitempty"`
URI string `json:"source_uri"`
OutputOnly bool `json:"source_uri_output_only,omitempty"`
}
SourceRunRecord is the payload for gonimbus.reflow.source.v1 JSONL records.
type StageStats ¶
type StageStats struct {
// contains filtered or unexported fields
}
StageStats accumulates sterile, process-local object-path stage timings for one reflow run. Safe for concurrent workers. Counts and nanoseconds only — no keys, URIs, paths, or provider identifiers.
Measure-first attribution for GON-067: not product throughput marketing.
func NewStageStats ¶
func NewStageStats() *StageStats
NewStageStats constructs an empty StageStats accumulator for a reflow run. Used by the CLI pool path; the engine constructs one inside prepareRun.
func (*StageStats) AttachLimiterPeaks ¶
func (s *StageStats) AttachLimiterPeaks(lim *ConcurrencyLimiter)
AttachLimiterPeaks copies per-domain peaks and dest-domain policy from the limiter into the stage-stats snapshot (called once at run end).
func (*StageStats) Snapshot ¶
func (s *StageStats) Snapshot() ObjectPathStageStatsRecord
Snapshot returns an immutable sterile record for JSONL emission.
type Summary ¶
type Summary struct {
SummaryRecord
}
Summary is the typed result of a reflow Run, mirroring the gonimbus.reflow.summary.v1 payload.
type SummaryRecord ¶
type SummaryRecord struct {
DestURI string `json:"dest_uri"`
DryRun bool `json:"dry_run"`
OnCollision string `json:"on_collision"`
ExecutionPath string `json:"execution_path"`
ConcurrencyStats
DestIfAbsentHonored *bool `json:"dest_ifabsent_honored"`
DestIfAbsentProbeStatus string `json:"dest_ifabsent_probe_status,omitempty"`
FallbackActive bool `json:"fallback_active"`
IfAbsentFallbackObjects int64 `json:"ifabsent_fallback_objects"`
Statuses map[string]int64 `json:"statuses,omitempty"`
Collisions map[string]int64 `json:"collisions,omitempty"`
InvalidInputs int64 `json:"invalid_inputs,omitempty"`
Errors int64 `json:"errors,omitempty"`
}
SummaryRecord is the payload for gonimbus.reflow.summary.v1 JSONL records.
Source Files
¶
- advisory.go
- checkpoint.go
- collision.go
- concurrency.go
- concurrency_fd_unix.go
- concurrency_memory_linux.go
- config.go
- destarbiter.go
- destination.go
- doc.go
- errors.go
- event.go
- execute.go
- execute_provenance.go
- ifabsent.go
- input.go
- layout.go
- memoryledger.go
- metadata.go
- provenance.go
- receipt.go
- records.go
- redaction.go
- runner.go
- runstats.go
- sanitize.go
- source.go
- stage_stats.go