observationpublication

package
v0.0.0-...-3d6ecf2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 25, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

Documentation

Overview

Package observationpublication publishes immutable, content-addressed source observations under one atomic repository current pointer.

Index

Constants

View Source 
const (
	MaxArchiveEntries   = 10_000_000
	MaxArchiveBytes     = int64(1 << 40)
	MaxArchivePathBytes = 512
)
View Source 
const (
	DownstreamAuthorityV1 = "observation-v1"
	DownstreamAuthorityV2 = "observation-v2"
)
View Source 
const (
	InventoryRootSchemaV2    = "phebs-observation-inventory-root-v2"
	InventorySegmentSchemaV2 = "phebs-observation-inventory-segment-v2"
	InventoryRootNameV2      = "inventory-root-v2.json"

	// Sixteen retained v1 record envelopes admit 4,000,000 records, more than
	// 15x the frozen T40.1 semantic profile's 262,144 distinct Go blobs. The
	// aggregate byte ceilings remain the shipped v1 generation ceilings.
	MaxInventorySegmentsV2     = 16
	MaxInventorySegmentRecords = MaxGenerationRecords
	MaxInventoryRecordsV2      = MaxInventorySegmentsV2 * MaxInventorySegmentRecords
	MaxInventoryRootBytesV2    = 1 << 20
	MaxInventorySegmentBytesV2 = MaxManifestBytes
)
View Source 
const (
	InventoryPublicationRootSchemaV2    = "phebs-observation-inventory-publication-root-v2"
	InventoryPublicationMarkerSchemaV2  = "phebs-observation-inventory-publication-marker-v2"
	InventoryPublicationDirectoryV2     = "v2"
	InventoryPublicationRootNameV2      = "current.json"
	InventoryPublicationMarkerNameV2    = "publishing.json"
	InventoryPublicationSourceNameV2    = "source"
	InventoryPublicationInventoryNameV2 = "inventory"

	MaxInventoryPublicationStagesV2  = 8
	MaxInventoryPublicationEntriesV2 = MaxRepositoryGenerations + MaxInventoryPublicationStagesV2 + 2
)
View Source 
const (
	InventoryScheduleStageV2            = "go-source-observation-inventory-v2"
	InventoryScheduleMaxAttemptsV2      = 5
	InventoryScheduleRepositoryTokensV2 = 1
)
View Source 
const (
	MaxLifecycleRepositories = 4_096
	MaxRepositoryGenerations = 64
	RetainedGenerations      = 2
	GenerationMaxAge         = 14 * 24 * time.Hour
)
View Source 
const (
	ManifestSchema = "phebs-observation-publication-v1"
	MemberSchema   = "phebs-observation-member-v1"
	RecordSchema   = "phebs-observation-record-v1"
	PointerSchema  = "phebs-observation-pointer-v1"
	MarkerSchema   = "phebs-observation-marker-v1"
	ReceiptSchema  = "phebs-observation-operation-receipt-v1"
	BindingSchema  = "phebs-observation-schedule-binding-v1"

	LanguagePack               = "go"
	MaxManifestBytes           = 8 << 20
	MaxMemberBytes       int64 = 128 << 20
	MaxObservationBytes        = 64 << 20
	MaxGenerationBytes   int64 = 20 << 30
	MaxGenerationRecords       = 250_000
	MaxArtifactNameBytes       = 255
)
View Source 
const (
	PlanningScheduleStage            = "go-source-observation-plan"
	PlanningScheduleMaxAttempts      = 5
	PlanningScheduleRepositoryTokens = 1
	PlanningBindingSchema            = "phebs-observation-planning-binding-v1"
)
View Source 
const (
	ProgressSchema   = "phebs-observation-progress-v1"
	ProgressSchemaV2 = "phebs-observation-progress-v2"
)
View Source 
const (
	ScheduleStage            = "go-source-observation"
	ScheduleChunkItems       = 8
	ScheduleMaxAttempts      = 5
	ScheduleRepositoryTokens = 2
)
View Source 
const CollectRemovalLimit = 8

CollectRemovalLimit bounds removals per owned planning run.

View Source 
const CollectScanLimit = 512

CollectScanLimit bounds the directory names one collection pass examines, and callers bound removals separately. Repeated bounded passes converge on a namespace that leaked before collection existed.

View Source 
const MaxInventoryLifecycleCandidatesV2 = 64

Variables

View Source 
var (
	ErrInvalid  = errors.New("invalid observation publication")
	ErrLimit    = errors.New("observation publication limit exceeded")
	ErrStale    = errors.New("observation publication is stale")
	ErrPinned   = errors.New("observation publication is pinned")
	ErrComplete = errors.New("observation publication is already complete")
)
View Source 
var ErrWorkUnavailable = errors.New("observation partition is unavailable")

Functions

func CurrentGeneration 

func CurrentGeneration(root, repository string) (string, error)

func GenerationDigest 

func GenerationDigest(partition sourcepartition.Manifest) (string, error)

func InventoryGenerationDigestV2 

func InventoryGenerationDigestV2(source sourcepartition.SuperRoot) (string, error)

func InventoryRootDigestV2 

func InventoryRootDigestV2(root InventoryRootV2) (string, error)

func ManifestDigest 

func ManifestDigest(value Manifest) (string, error)

func Recover 

func Recover(ctx context.Context, root, repository string) (bool, error)

Recover resolves one marker without weakening the prior pointer. A complete staged generation is activated; an incomplete stage is removed and the prior current publication remains authoritative.

func RecoverInventoryPublicationV2 

func RecoverInventoryPublicationV2(ctx context.Context, root, repository string) (bool, error)

RecoverInventoryPublicationV2 is the startup-only completion path. The caller holds the common backup/publication mutation lock before workers are admitted. Incomplete or corrupt stage bytes remain non-authoritative and the prior root is unchanged.

func RestoreArchive 

func RestoreArchive(ctx context.Context, archivePath, root string) error

func RestoreArchiveWithStage 

func RestoreArchiveWithStage(ctx context.Context, archivePath, root, stage string) error

RestoreArchiveWithStage restores through an explicit sibling stage. The recovery workflow uses a stage outside its data directory so interruption leaves that directory empty and a subsequent invocation can resume. The stage and destination must be disjoint.

func ValidateInventoryRootV2 

func ValidateInventoryRootV2(root InventoryRootV2) error

func ValidateInventorySourceV2 

func ValidateInventorySourceV2(
	directory string, root InventoryRootV2, plan *sourcepartition.SuperPlan,
) error

ValidateInventorySourceV2 proves that every observation member binds the exact ordered member of the complete source super-root named by the v2 root. It reads only root/segment controls; source and observation member bytes are already transitively bound by their digests.

func ValidateInventoryStageV2 

func ValidateInventoryStageV2(ctx context.Context, directory string, expected InventoryRootV2) error

func ValidateManifest 

func ValidateManifest(value Manifest) error

ValidateManifest exposes the closed publication-control validator to later immutable consumers without granting filesystem or mutation capabilities.

func ValidateProgress 

func ValidateProgress(progress Progress) error

Types

type ArchiveReport 

type ArchiveReport struct {
	Publications        int
	V1Publications      int
	V2Publications      int
	Files               int
	Bytes               int64
	Omitted             int
	OmittedPublications int
	OmittedArtifacts    int
	StaleMarkers        int
}

func CreateArchive 

func CreateArchive(ctx context.Context, root, output string) (ArchiveReport, error)

CreateArchive includes only fully validated current publications. A corrupt or marker-covered derived generation is omitted rather than blocking backup of precious state.

func VerifyArchive 

func VerifyArchive(ctx context.Context, archivePath string) (ArchiveReport, error)

type Cache 

type Cache struct {
	// contains filtered or unexported fields
}

func (*Cache) Acquire 

func (cache *Cache) Acquire(ctx context.Context, root, repository string) (*Lease, error)

func (*Cache) AcquireGeneration 

func (cache *Cache) AcquireGeneration(
	ctx context.Context, root, repository, generation string,
) (*Lease, error)

AcquireGeneration pins and fully validates one exact historical generation. The provisional cache entry is visible before filesystem access so a lifecycle sweep that starts later observes the pin. A sweep that already renamed the generation makes the strict open fail instead.

func (*Cache) Pinned 

func (cache *Cache) Pinned(repository, generation string) bool

func (*Cache) Retire 

func (cache *Cache) Retire(repository string)

type CollectResult 

type CollectResult struct {
	Plans    int
	Stages   int
	Bindings int
	Aborted  bool
}

type DownstreamAuthority 

type DownstreamAuthority struct {
	Version                     string `json:"version"`
	Repository                  string `json:"repository"`
	SourceGenerationDigest      string `json:"source_generation_digest"`
	SourceRootDigest            string `json:"source_root_digest"`
	ObservationGenerationDigest string `json:"observation_generation_digest"`
	ObservationRootDigest       string `json:"observation_root_digest"`
	PartitionPolicyDigest       string `json:"partition_policy_digest"`
	ObservationPolicyDigest     string `json:"observation_policy_digest"`
	InventoryPolicyDigest       string `json:"inventory_policy_digest,omitempty"`
	RecordCount                 int    `json:"record_count"`
	ObservedCount               int    `json:"observed_count"`
}

DownstreamAuthority is the small, versioned observation identity used by derived generations. V1 fields retain the shipped manifest identity; V2 names the independently published source super-root and inventory root. Consumers must bind the complete value rather than treating a generation digest as a sufficient substitute for its policy and root digests.

func CurrentInventoryDownstreamAuthorityV2 

func CurrentInventoryDownstreamAuthorityV2(
	ctx context.Context, root, repository string,
) (DownstreamAuthority, error)

CurrentInventoryDownstreamAuthorityV2 reads one pointer-sized publication reference, the selected at-most-1-MiB inventory root, and the pointer again. It performs no segment/member/source/observation scan.

type DownstreamSource 

type DownstreamSource interface {
	DownstreamAuthority() DownstreamAuthority
	WalkObserved(context.Context, func(Record, sourceobservation.Observation) error) error
}

DownstreamSource is the version-neutral input surface for relationship component builders. WalkObserved is cold construction work; Authority is a bounded control read used by planning and no-op comparison.

type InventoryAuthorityV2 

type InventoryAuthorityV2 struct {
	SourceGenerationDigest      string
	ObservationGenerationDigest string
	SourceRootDigest            string
	InventoryRootDigest         string
}

InventoryAuthorityV2 is the small control-only authority T40.10 binds into each extraction plan. It contains no observation or source member bytes.

func CurrentInventoryAuthorityReferenceV2 

func CurrentInventoryAuthorityReferenceV2(
	ctx context.Context,
	root,
	repository string,
) (InventoryAuthorityV2, error)

CurrentInventoryAuthorityReferenceV2 performs the final publication-fence read: one latest pointer plus its selected small source super-root, followed by a pointer reread. The generation was fully validated when installed and by CurrentInventoryAuthorityV2 during planning, so the final fence does not rescan every source/inventory segment for every completed domain.

func CurrentInventoryAuthorityV2 

func CurrentInventoryAuthorityV2(
	ctx context.Context,
	root,
	repository string,
) (InventoryAuthorityV2, error)

CurrentInventoryAuthorityV2 validates the selected source/inventory roots and rereads the latest pointer before returning. Callers performing a final publication fence hold the observation transition lock around this call.

type InventoryBuildRequestV2 

type InventoryBuildRequestV2 struct {
	OutputDirectory     string
	RepositoryDirectory string
	Plan                *sourcepartition.SuperPlan
	PriorDirectory      string
	Metrics             *InventoryMetricsV2
}

type InventoryCacheV2 

type InventoryCacheV2 struct {
	// contains filtered or unexported fields
}

InventoryCacheV2 validates an immutable stage once per exact root digest. Warm leases perform no segment, member, or observation reread; T40.6 may use Pinned when it gives these non-authoritative stages lifecycle ownership.

func (*InventoryCacheV2) Acquire 

func (cache *InventoryCacheV2) Acquire(
	ctx context.Context, directory string, expected InventoryRootV2,
) (*InventoryLeaseV2, error)

func (*InventoryCacheV2) AcquireCurrent 

func (cache *InventoryCacheV2) AcquireCurrent(
	ctx context.Context, root, repository string, expected DownstreamAuthority,
) (*InventoryLeaseV2, error)

AcquireCurrent pins and fully validates the exact V2 inventory selected by a shallow planning authority. A pointer reread prevents a worker from accepting a valid historical generation after current authority moves.

func (*InventoryCacheV2) Pinned 

func (cache *InventoryCacheV2) Pinned(directory, digestValue string) bool

func (*InventoryCacheV2) PinnedGeneration 

func (cache *InventoryCacheV2) PinnedGeneration(repository, generation string) bool

PinnedGeneration is the lifecycle-facing identity lookup. Cache keys remain exact directory/root-digest pairs, but lifecycle authority is the bound repository/generation pair stored inside the validated publication.

type InventoryDurablePinCheckerV2 

type InventoryDurablePinCheckerV2 interface {
	PinnedInventoryProofV2(context.Context, string, string) (bool, error)
	PinnedInventoryInvestigationV2(context.Context, string, string) (bool, error)
}

type InventoryGenerationRefV2 

type InventoryGenerationRefV2 struct {
	GenerationDigest  string `json:"generation_digest"`
	Directory         string `json:"directory"`
	SourceRootDigest  string `json:"source_root_digest"`
	InventoryDigest   string `json:"inventory_digest"`
	SourceSegments    int    `json:"source_segments"`
	InventorySegments int    `json:"inventory_segments"`
	Records           int    `json:"records"`
	EncodedBytes      int64  `json:"encoded_bytes"`
	ObservationBytes  int64  `json:"observation_bytes"`
}

InventoryGenerationRefV2 binds the independently validated T40.3 source super-root and T40.4 observation inventory installed as one immutable generation. The small reference is lifecycle and archive authority; corpus bytes remain in the digest-named directory.

type InventoryLeaseV2 

type InventoryLeaseV2 struct {
	// contains filtered or unexported fields
}

func (*InventoryLeaseV2) Publication 

func (lease *InventoryLeaseV2) Publication() *InventoryPublicationV2

func (*InventoryLeaseV2) Release 

func (lease *InventoryLeaseV2) Release()

type InventoryLifecycleResultV2 

type InventoryLifecycleResultV2 struct {
	Cursor                  string
	Scanned                 int
	Deleted                 int
	More                    bool
	CurrentRecords          int
	CurrentEncodedBytes     int64
	CurrentObservationBytes int64
}

func SweepInventoryLifecycleV2 

func SweepInventoryLifecycleV2(
	ctx context.Context, root string, now time.Time, cursor string,
	pins InventoryPinCheckerV2, candidateLimit, deleteLimit int,
) (InventoryLifecycleResultV2, error)

SweepInventoryLifecycleV2 examines one repository's joint source/ observation namespace. Current, rollback-floor, marker stage, active lease, proof, and Investigation roots outrank collection. A candidate is renamed out of the publication namespace before bounded deletion, and interrupted collecting/stage drains resume on later fair turns.

type InventoryMemberV2 

type InventoryMemberV2 struct {
	Ordinal            int    `json:"ordinal"`
	Count              int    `json:"count"`
	Name               string `json:"name"`
	SourceSegment      int    `json:"source_segment"`
	SourceOrdinal      int    `json:"source_ordinal"`
	SourceMemberDigest string `json:"source_member_digest"`
	SourcePrefix       string `json:"source_prefix"`
	SourcePrefixBits   int    `json:"source_prefix_bits"`
	RecordCount        int    `json:"record_count"`
	ObservedCount      int    `json:"observed_count"`
	UnsupportedCount   int    `json:"unsupported_count"`
	ContentBytes       int64  `json:"content_bytes"`
	Digest             string `json:"digest"`
}

type InventoryMetricsV2 

type InventoryMetricsV2 struct {
	ReadBlobs          int
	ParsedBlobs        int
	ReusedMembers      int
	ReusedObservations int
	WrittenMemberBytes int64
	LinkedMemberBytes  int64
}

type InventoryPinCheckerV2 

type InventoryPinCheckerV2 interface {
	PinnedInventoryV2(context.Context, string, string) (bool, error)
}

InventoryPinCheckerV2 combines active reader leases with durable proof and Investigation ownership. The lifecycle owner invokes it only after exact root/marker revalidation while holding the common mutation/backup lock.

type InventoryPinsV2 

type InventoryPinsV2 struct {
	Cache   *InventoryCacheV2
	Durable InventoryDurablePinCheckerV2
}

func (InventoryPinsV2) PinnedInventoryV2 

func (pins InventoryPinsV2) PinnedInventoryV2(
	ctx context.Context, repository, generation string,
) (bool, error)

type InventoryPolicyV2 

type InventoryPolicyV2 struct {
	Name                     string `json:"name"`
	MaxSegments              int    `json:"max_segments"`
	MaxSegmentRecords        int    `json:"max_segment_records"`
	MaxAggregateRecords      int    `json:"max_aggregate_records"`
	MaxMemberBytes           int64  `json:"max_member_bytes"`
	MaxAggregateEncodedBytes int64  `json:"max_aggregate_encoded_bytes"`
	MaxAggregateObjectBytes  int64  `json:"max_aggregate_object_bytes"`
	ObjectByteKind           string `json:"object_byte_kind"`
}

type InventoryPublicationMarkerV2 

type InventoryPublicationMarkerV2 struct {
	Schema       string                      `json:"schema"`
	Repository   string                      `json:"repository"`
	TransitionID string                      `json:"transition_id"`
	Stage        string                      `json:"stage"`
	Candidate    *InventoryGenerationRefV2   `json:"candidate,omitempty"`
	Previous     *InventoryPublicationRootV2 `json:"previous,omitempty"`
}

type InventoryPublicationRootV2 

type InventoryPublicationRootV2 struct {
	Schema     string                    `json:"schema"`
	Repository string                    `json:"repository"`
	Current    InventoryGenerationRefV2  `json:"current"`
	Prior      *InventoryGenerationRefV2 `json:"prior,omitempty"`
}

func CompleteInventoryPublicationV2 

func CompleteInventoryPublicationV2(
	ctx context.Context, root, repository, transitionID string,
	fence func(context.Context) error,
) (InventoryPublicationRootV2, error)

CompleteInventoryPublicationV2 validates every source and observation byte before the caller-provided durable worker fence. The fence and exact marker token are rechecked before rename/root mutation, so a stale worker cannot publish even when it finished valid content-addressed bytes.

func ReadInventoryPublicationRootV2 

func ReadInventoryPublicationRootV2(root, repository string) (InventoryPublicationRootV2, error)

type InventoryPublicationTransitionV2 

type InventoryPublicationTransitionV2 struct {
	Repository         string
	TransitionID       string
	Directory          string
	SourceDirectory    string
	InventoryDirectory string
}

func BeginInventoryPublicationV2 

func BeginInventoryPublicationV2(root, repository string) (InventoryPublicationTransitionV2, error)

BeginInventoryPublicationV2 creates or resumes one marker-owned stage. A worker builds the complete source super-root under SourceDirectory and the bound observation inventory under InventoryDirectory. The marker exists before corpus work, so a restart can validate and complete only an exact finished pair while preserving the prior root for every incomplete prefix.

func RestartInventoryPublicationV2 

func RestartInventoryPublicationV2(
	ctx context.Context, root, repository, transitionID string,
	fence func(context.Context) error,
) (InventoryPublicationTransitionV2, error)

RestartInventoryPublicationV2 retires one incomplete marker-owned stage without recursive deletion. A later lifecycle turn drains the collecting name within its deletion budget, while a fresh transition can resume work from a clean source/observation pair. The durable worker fence prevents a stale process from retiring a replacement worker's stage.

type InventoryPublicationV2 

type InventoryPublicationV2 struct {
	// contains filtered or unexported fields
}

func OpenInventoryV2 

func OpenInventoryV2(ctx context.Context, directory string, expected InventoryRootV2) (*InventoryPublicationV2, error)

func OpenInventoryV2Keyed 

func OpenInventoryV2Keyed(directory, repository string) (*InventoryPublicationV2, error)

func (*InventoryPublicationV2) DownstreamAuthority 

func (publication *InventoryPublicationV2) DownstreamAuthority() DownstreamAuthority

func (*InventoryPublicationV2) Lookup 

func (publication *InventoryPublicationV2) Lookup(ctx context.Context, objectID string) (Record, *sourceobservation.Observation, error)

func (*InventoryPublicationV2) Root 

func (publication *InventoryPublicationV2) Root() InventoryRootV2

func (*InventoryPublicationV2) WalkObserved 

func (publication *InventoryPublicationV2) WalkObserved(
	ctx context.Context,
	visit func(Record, sourceobservation.Observation) error,
) error

WalkObserved streams a validated V2 inventory in canonical segment/member order. It never opens the independently retained source super-root: every member is already bound to that root by the validated inventory controls.

type InventoryRecoveryReportV2 

type InventoryRecoveryReportV2 struct {
	Repositories int
	Completed    int
	Incomplete   int
}

func RecoverInventoryPublicationsV2 

func RecoverInventoryPublicationsV2(ctx context.Context, root string) (InventoryRecoveryReportV2, error)

type InventoryRootV2 

type InventoryRootV2 struct {
	Schema                    string                    `json:"schema"`
	Repository                string                    `json:"repository"`
	Language                  string                    `json:"language"`
	SourceGenerationDigest    string                    `json:"source_generation_digest"`
	SourcePartitionGeneration string                    `json:"source_partition_generation"`
	SourcePartitionRootDigest string                    `json:"source_partition_root_digest"`
	PartitionPolicyDigest     string                    `json:"partition_policy_digest"`
	ObservationPolicyDigest   string                    `json:"observation_policy_digest"`
	InventoryPolicy           InventoryPolicyV2         `json:"inventory_policy"`
	InventoryPolicyDigest     string                    `json:"inventory_policy_digest"`
	GenerationDigest          string                    `json:"generation_digest"`
	Segments                  []InventorySegmentEntryV2 `json:"segments"`
	MemberCount               int                       `json:"member_count"`
	RecordCount               int                       `json:"record_count"`
	ObservedCount             int                       `json:"observed_count"`
	UnsupportedCount          int                       `json:"unsupported_count"`
	EncodedMemberBytes        int64                     `json:"encoded_member_bytes"`
	ObservationCount          int                       `json:"observation_count"`
	ObservationBytes          int64                     `json:"observation_bytes"`
	OperationReceipt          OperationReceipt          `json:"operation_receipt"`
	Digest                    string                    `json:"digest"`
}

func BuildInventoryStageV2 

func BuildInventoryStageV2(ctx context.Context, request InventoryBuildRequestV2) (InventoryRootV2, error)

func ReadInventoryRootV2 

func ReadInventoryRootV2(directory, repository string) (InventoryRootV2, error)

type InventorySegmentEntryV2 

type InventorySegmentEntryV2 struct {
	Ordinal            int    `json:"ordinal"`
	Count              int    `json:"count"`
	Directory          string `json:"directory"`
	SegmentDigest      string `json:"segment_digest"`
	MemberCount        int    `json:"member_count"`
	FirstPrefix        string `json:"first_prefix"`
	LastPrefix         string `json:"last_prefix"`
	RecordCount        int    `json:"record_count"`
	ObservedCount      int    `json:"observed_count"`
	UnsupportedCount   int    `json:"unsupported_count"`
	EncodedMemberBytes int64  `json:"encoded_member_bytes"`
	ObservationCount   int    `json:"observation_count"`
	ObservationBytes   int64  `json:"observation_bytes"`
}

type InventorySegmentV2 

type InventorySegmentV2 struct {
	Schema                    string              `json:"schema"`
	Repository                string              `json:"repository"`
	GenerationDigest          string              `json:"generation_digest"`
	SourcePartitionRootDigest string              `json:"source_partition_root_digest"`
	ObservationPolicyDigest   string              `json:"observation_policy_digest"`
	Ordinal                   int                 `json:"ordinal"`
	Count                     int                 `json:"count"`
	Members                   []InventoryMemberV2 `json:"members"`
	RecordCount               int                 `json:"record_count"`
	ObservedCount             int                 `json:"observed_count"`
	UnsupportedCount          int                 `json:"unsupported_count"`
	EncodedMemberBytes        int64               `json:"encoded_member_bytes"`
	ObservationCount          int                 `json:"observation_count"`
	ObservationBytes          int64               `json:"observation_bytes"`
	OperationReceipt          OperationReceipt    `json:"operation_receipt"`
	Digest                    string              `json:"digest"`
}

type Lease 

type Lease struct {
	// contains filtered or unexported fields
}

func (*Lease) Publication 

func (lease *Lease) Publication() *Publication

func (*Lease) Release 

func (lease *Lease) Release()

type LifecycleResult 

type LifecycleResult struct {
	Cursor  string
	Scanned int
	Deleted int
	More    bool
}

func SweepLifecycle 

func SweepLifecycle(
	ctx context.Context, root string, now time.Time, cursor string,
	pins PinChecker, deleteLimit int,
) (LifecycleResult, error)

SweepLifecycle examines one repository and removes entries from at most one unrooted generation. The caller must hold the shared backup/publication mutation lock. A generation is first renamed out of the publication namespace and is then drained with at most deleteLimit filesystem removals per call.

type Manifest 

type Manifest struct {
	Schema                    string            `json:"schema"`
	Repository                string            `json:"repository"`
	Language                  string            `json:"language"`
	SourceGenerationDigest    string            `json:"source_generation_digest"`
	PartitionGenerationDigest string            `json:"partition_generation_digest"`
	PartitionManifestDigest   string            `json:"partition_manifest_digest"`
	PartitionPolicyDigest     string            `json:"partition_policy_digest"`
	ObservationPolicyDigest   string            `json:"observation_policy_digest"`
	GenerationDigest          string            `json:"generation_digest"`
	Members                   []Member          `json:"members"`
	RecordCount               int               `json:"record_count"`
	ObservedCount             int               `json:"observed_count"`
	UnsupportedCount          int               `json:"unsupported_count"`
	EncodedMemberBytes        int64             `json:"encoded_member_bytes"`
	ObservationBytes          int64             `json:"observation_bytes"`
	OperationReceipt          *OperationReceipt `json:"operation_receipt,omitempty"`
	Digest                    string            `json:"digest"`
}

type Marker 

type Marker struct {
	Schema           string `json:"schema"`
	Repository       string `json:"repository"`
	GenerationDigest string `json:"generation_digest"`
	ManifestDigest   string `json:"manifest_digest,omitempty"`
}

type Member 

type Member struct {
	Ordinal            int    `json:"ordinal"`
	Count              int    `json:"count"`
	Name               string `json:"name"`
	SourceMemberDigest string `json:"source_member_digest"`
	RecordCount        int    `json:"record_count"`
	ObservedCount      int    `json:"observed_count"`
	UnsupportedCount   int    `json:"unsupported_count"`
	ContentBytes       int64  `json:"content_bytes"`
	Digest             string `json:"digest"`
}

type Metrics 

type Metrics struct {
	ReadBlobs          int
	ParsedBlobs        int
	ReusedObservations int
	UnsupportedBlobs   int
	WrittenBytes       int64
}

type OperationReceipt 

type OperationReceipt struct {
	Schema             string        `json:"schema"`
	InputBlobs         int           `json:"input_blobs"`
	SourceBlobReads    int           `json:"source_blob_reads"`
	ParsedObservations int           `json:"parsed_observations"`
	ReusedObservations int           `json:"reused_observations"`
	ObservedBlobs      int           `json:"observed_blobs"`
	UnsupportedBlobs   int           `json:"unsupported_blobs"`
	UnsupportedReasons []ReasonCount `json:"unsupported_reasons"`
}

OperationReceipt is the source-free, digest-bound accounting for one complete publication. SourceBlobReads counts inputs incorporated into successfully published members; failed or interrupted attempts are outside this immutable publication receipt and remain scheduler diagnostics.

type PinChecker 

type PinChecker interface {
	Pinned(repository, generation string) bool
}

type PlanningEnqueue 

type PlanningEnqueue string

PlanningEnqueue is the closed, source-free result of one bounded request. It is operational diagnostics only; it does not establish publication. 

const (
	PlanningCurrent  PlanningEnqueue = "current"
	PlanningActive   PlanningEnqueue = "active"
	PlanningFailed   PlanningEnqueue = "failed"
	PlanningEnqueued PlanningEnqueue = "enqueued"
)

type PlanningProgress 

type PlanningProgress struct {
	State                  string                   `json:"state"` // active | settled | failed
	ScheduleGeneration     string                   `json:"schedule_generation"`
	TargetGeneration       string                   `json:"target_generation"`
	SourceGenerationDigest string                   `json:"source_generation_digest"`
	Pending                int                      `json:"pending"`
	Running                int                      `json:"running"`
	Succeeded              int                      `json:"succeeded"`
	Failed                 int                      `json:"failed"`
	Refusal                *pipelinerefusal.Receipt `json:"refusal,omitempty"`
}

type Pointer 

type Pointer struct {
	Schema           string `json:"schema"`
	Repository       string `json:"repository"`
	GenerationDigest string `json:"generation_digest"`
	ManifestDigest   string `json:"manifest_digest"`
	ManifestName     string `json:"manifest_name"`
}

type Progress 

type Progress struct {
	SchemaVersion          string               `json:"schema"`
	SelectedVersion        string               `json:"selected_version,omitempty"`
	Repository             string               `json:"repository"`
	State                  string               `json:"state"` // current | building | failed | stale | unavailable
	SourceGenerationDigest string               `json:"source_generation_digest"`
	Publication            *PublicationProgress `json:"publication,omitempty"`
	Planning               *PlanningProgress    `json:"planning,omitempty"`
	Schedule               *ScheduleProgress    `json:"schedule,omitempty"`
}

type ProgressFailureStore 

type ProgressFailureStore interface {
	GetGenerationScheduleFailure(
		context.Context, string, string, string,
	) (*store.GenerationScheduleFailure, error)
}

type ProgressReadError 

type ProgressReadError struct {
	Stage ProgressReadStage
	Err   error
}

ProgressReadError retains one closed source-free failure stage while preserving the typed cause for boundary classification. It never replaces an ErrStale, context, store, or immutable-corruption cause.

func (*ProgressReadError) Error 

func (failure *ProgressReadError) Error() string

func (*ProgressReadError) Unwrap 

func (failure *ProgressReadError) Unwrap() error

type ProgressReadStage 

type ProgressReadStage string
const (
	ProgressReadStageControl     ProgressReadStage = "control"
	ProgressReadStagePublication ProgressReadStage = "publication"
	ProgressReadStagePlanning    ProgressReadStage = "planning"
	ProgressReadStageSchedule    ProgressReadStage = "schedule"
	ProgressReadStageProjection  ProgressReadStage = "projection"
)

type ProgressReader 

type ProgressReader struct {
	DataDir     string
	Store       ProgressStore
	Cache       *Cache
	InventoryV2 bool
}

ProgressReader composes one fully validated cached publication with bounded source, marker, plan, and schedule controls. It never reads source blobs or observation members on a warm cache hit.

func (*ProgressReader) Read 

func (reader *ProgressReader) Read(ctx context.Context, repository string) (Progress, error)

type ProgressStore 

type ProgressStore interface {
	GetGenerationSchedule(context.Context, string, string) (*store.GenerationSchedule, error)
}

type Publication 

type Publication struct {
	// contains filtered or unexported fields
}

func Open 

func Open(ctx context.Context, root, repository string) (*Publication, error)

func Publish 

func Publish(
	ctx context.Context, root, repositoryDirectory string,
	plan *sourcepartition.Plan, metrics *Metrics,
) (*Publication, error)

Publish is the bounded single-process convenience boundary. Durable workers may call Begin, BuildPartition, and Finalize separately for the same bytes.

func (*Publication) DownstreamAuthority 

func (publication *Publication) DownstreamAuthority() DownstreamAuthority

func (*Publication) Manifest 

func (publication *Publication) Manifest() Manifest

func (*Publication) WalkObserved 

func (publication *Publication) WalkObserved(
	ctx context.Context,
	visit func(Record, sourceobservation.Observation) error,
) error

WalkObserved rereads the immutable member and observation bytes owned by an already validated publication and yields each observed content identity once with all of its placements. Unsupported source records are not observations and remain accounted by the manifest operation receipt.

type PublicationProgress 

type PublicationProgress struct {
	State                  string            `json:"state"` // current | stale
	GenerationDigest       string            `json:"generation_digest"`
	SourceGenerationDigest string            `json:"source_generation_digest"`
	RecordCount            int               `json:"record_count"`
	ObservedCount          int               `json:"observed_count"`
	UnsupportedCount       int               `json:"unsupported_count"`
	ReceiptState           string            `json:"receipt_state"` // complete | legacy_unavailable
	Receipt                *OperationReceipt `json:"receipt,omitempty"`
	SourceSegments         int               `json:"source_segments,omitempty"`
	InventorySegments      int               `json:"inventory_segments,omitempty"`
	EncodedMemberBytes     int64             `json:"encoded_member_bytes,omitempty"`
	ObservationBytes       int64             `json:"observation_bytes,omitempty"`
}

type ReasonCount 

type ReasonCount struct {
	Reason string `json:"reason"`
	Count  int    `json:"count"`
}

ReasonCount is one closed unsupported-source classification. It contains no path, object identity, source sample, or parser error text.

type Record 

type Record struct {
	Schema            string                      `json:"schema"`
	ObjectID          string                      `json:"object_id"`
	DeclaredBytes     int64                       `json:"declared_bytes"`
	Placements        []sourcepartition.Placement `json:"placements"`
	State             string                      `json:"state"` // observed | unsupported
	Reason            string                      `json:"reason,omitempty"`
	ContentDigest     string                      `json:"content_digest"`
	ObservationDigest string                      `json:"observation_digest,omitempty"`
	ObservationName   string                      `json:"observation_name,omitempty"`
	ObservationOrigin string                      `json:"observation_origin,omitempty"` // parsed | reused
}

type Runtime 

type Runtime struct {
	DataDir string
	Store   store.GenerationSchedulerStore
	// Cache pins a prevalidated historical generation while planning performs
	// the short final reactivation fence. It is required by EnqueuePlanning and
	// HandlePlanning; legacy v1 Reconcile/Handle do not depend on it.
	Cache *Cache
	// AcquireTransition excludes lifecycle/index publication mutation for the
	// short final source/schedule re-fence and pointer/ownership transition.
	AcquireTransition func(context.Context) (func(), error)
	Admit             func(context.Context) error
	// InventoryV2 enables the durable source-super-root and observation-inventory
	// stage. It is explicit so historical/runtime fixtures retain the v1
	// contract unless they opt into the production pipeline.
	InventoryV2 bool
	// OnPublished runs after an exact current observation is present. A
	// failure keeps the scheduler chunk retryable without rolling back the
	// already-complete content-addressed observation publication.
	OnPublished func(context.Context, string) error
	// contains filtered or unexported fields
}

func (*Runtime) CollectSupersededPlans 

func (runtime *Runtime) CollectSupersededPlans(
	ctx context.Context, chunk store.GenerationChunk, limit int,
) (CollectResult, error)

CollectSupersededPlans removes plan directories and schedule/planning bindings that no current authority names: not the active or settled current schedules, not a recovery identity a concurrent enqueue may be minting, not the current publication pointer or marker, and not an in-process build. It renews the exact planning lease inside the destructive namespace fence and re-reads every authority there, aborting on any change.

func (*Runtime) EnqueuePlanning 

func (runtime *Runtime) EnqueuePlanning(
	ctx context.Context, repository string,
) (PlanningEnqueue, error)

EnqueuePlanning gives the current source generation durable planning ownership. It reads bounded controls only; source members and Git objects are opened exclusively by HandlePlanning after ownership exists.

func (*Runtime) Handle 

func (runtime *Runtime) Handle(ctx context.Context, chunk store.GenerationChunk) error

Handle executes one durable scheduler chunk. Publication is attempted only after the final member exists. A slower earlier chunk also observes that sentinel and can finalize, so correctness does not depend on completion order.

func (*Runtime) HandleInventoryV2 

func (runtime *Runtime) HandleInventoryV2(ctx context.Context, chunk store.GenerationChunk) error

HandleInventoryV2 builds the joint v2 authority outside the mutation fence, then renews the exact scheduler lease and installs it while holding the same exclusive boundary used by index, lifecycle, backup, and relationship work.

func (*Runtime) HandlePlanning 

func (runtime *Runtime) HandlePlanning(ctx context.Context, chunk store.GenerationChunk) error

HandlePlanning performs the formerly synchronous source-partition census. The short final fence, not the long census, holds the repository mirror lock.

func (*Runtime) Reconcile 

func (runtime *Runtime) Reconcile(ctx context.Context, repository string) error

Reconcile binds the current complete repository source generation to one durable observation schedule. Callers serialize it with repository index publication; the normal index completion hook already holds that lock.

type ScheduleProgress 

type ScheduleProgress struct {
	State                 string `json:"state"` // active | settled | failed
	ScheduleGeneration    string `json:"schedule_generation"`
	PublicationGeneration string `json:"publication_generation"`
	TotalPartitions       int    `json:"total_partitions"`
	Materialized          int    `json:"materialized"`
	Pending               int    `json:"pending"`
	Running               int    `json:"running"`
	Succeeded             int    `json:"succeeded"`
	Failed                int    `json:"failed"`
}

type Stage 

type Stage struct {
	// contains filtered or unexported fields
}

func Begin 

func Begin(root string, partition sourcepartition.Manifest) (*Stage, error)

func (*Stage) BuildPartition 

func (stage *Stage) BuildPartition(
	ctx context.Context, plan *sourcepartition.Plan, repositoryDirectory string,
	ordinal int, metrics *Metrics,
) error

func (*Stage) Finalize 

func (stage *Stage) Finalize(ctx context.Context) (*Publication, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.