relationshippublication

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: 35 Imported by: 0

Documentation

Overview

Package relationshippublication projects repository-shared relationship postings onto exact service-catalog placement authority and publishes one atomic repository root with independent service partitions.

Index

Constants

View Source 
const (
	MaxArchiveEntries   = 20_000_000
	MaxArchiveBytes     = int64(1 << 40)
	MaxArchivePathBytes = 512
)
View Source 
const (
	MaxLifecycleRepositories = 4_096
	MaxRepositoryGenerations = 64
	// MaxRepositoryRepairEntries is a bounded emergency envelope above the
	// steady-state generation ceiling. It admits orphan stages and excess
	// generations so lifecycle/recovery can drain them instead of wedging.
	MaxRepositoryRepairEntries = 20_000
	MaxStageRepairFiles        = 20_000
	RetainedGenerations        = 2
	GenerationMaxAge           = 14 * 24 * time.Hour
)
View Source 
const (
	RootSchema               = "phebs-relationship-root-v1"
	RootSchemaV2             = "phebs-relationship-root-v2"
	RepositoryMemberSchema   = "phebs-relationship-repository-member-v1"
	RepositoryMemberSchemaV2 = "phebs-relationship-repository-member-v2"
	ProjectionSchema         = "phebs-relationship-projection-v1"
	ServiceMemberSchema      = "phebs-service-relationship-member-v1"
	ServiceMemberSchemaV2    = "phebs-service-relationship-member-v2"
	ServiceReferenceSchema   = "phebs-service-relationship-reference-v1"
	PolicySchema             = "phebs-relationship-policy-v1"
	PointerSchema            = "phebs-relationship-pointer-v1"
	MarkerSchema             = "phebs-relationship-marker-v1"

	RepositoryBuckets               = 256
	MaxServices                     = 4_000
	MaxProjectionRecords            = 2_000_000
	MaxServiceReferences            = 1_000_000
	MaxTotalServiceReferences       = 20_000_000
	MaxClaimsPerPlacement           = 4_000
	MaxRolesPerClaim                = 5
	MaxTextBytes                    = 4_096
	MaxProjectionBytes              = 1 << 20
	MaxReferenceBytes               = 64 << 10
	MaxRepositoryMemberBytes        = 128 << 20
	MaxServiceMemberBytes           = 128 << 20
	MaxRootBytes                    = 8 << 20
	MaxGenerationBytes        int64 = 20 << 30
	// MaxResidentChargeBytes is a conservative in-process admission fence.
	// Builders charge canonical bytes plus fixed map/slice overhead before
	// retaining each repository projection or service reference. This keeps
	// the registered one-GiB scheduler class honest without pretending that
	// the encoded-generation ceiling is a resident-memory promise.
	MaxResidentChargeBytes int64 = 512 << 20
)
View Source 
const (
	// MaxRecoveryWork is one coherent startup bound across namespace discovery,
	// directory scans, and repair deletions. It admits 65 entries in every
	// supported repository/component namespace without admitting the full
	// 82-million-entry Cartesian repair envelope.
	MaxRecoveryWork       = 2_000_000
	MaxRecoveryOwners     = MaxLifecycleRepositories * (RetainedGenerations + 2)
	MaxRecoveryNamespaces = MaxLifecycleRepositories * 4
)
View Source 
const (
	ScheduleStage            = "service-relationship"
	ScheduleMaxAttempts      = 5
	ScheduleRepositoryTokens = 1
	BindingSchema            = "phebs-relationship-schedule-binding-v1"
	BindingSchemaV2          = "phebs-relationship-schedule-binding-v2"
	BindingSchemaV3          = "phebs-relationship-schedule-binding-v3"
	ResolverResidentLimit    = 128 << 20
	RPCResidentLimit         = 192 << 20
	KafkaResidentLimit       = 160 << 20
)
View Source 
const (
	UnavailableSchema = "phebs-relationship-unavailable-v1"
	UnavailableName   = "unavailable.json"
)
View Source 
const CitationContentLimit = sourceobservation.MaxSourceBytes

Variables

View Source 
var (
	ErrInvalid    = errors.New("invalid relationship publication")
	ErrLimit      = errors.New("relationship publication bound exceeded")
	ErrNotFound   = errors.New("relationship publication not found")
	ErrPublishing = errors.New("relationship publication changed")
	ErrDenied     = errors.New("relationship publication not found")
)
View Source 
var ErrServiceUnavailable = errors.New("service relationship partition unavailable")

Functions

func ConfirmLifecycleUnpin 

func ConfirmLifecycleUnpin(
	ctx context.Context, dataDir, repositoryHashValue, owner string,
) error

ConfirmLifecycleUnpin durably records that the store owner was removed. Until this marker exists, SweepLifecycle retains the collecting directory and returns the same owner identity for an idempotent retry.

func ConfirmUnavailable 

func ConfirmUnavailable(
	ctx context.Context, root, repository string, expected *Unavailable,
) error

ConfirmUnavailable is the result-time fence for a current-reader gap. A nil expected marker proves that the repository remains truly absent; a non-nil marker proves the exact unavailable authority and digest remain selected.

func OpenAuthorizedRoot 

func OpenAuthorizedRoot(
	ctx context.Context,
	root, repository string,
	authorize AuthorizeRepository,
) (*Publication, Root, error)

OpenAuthorizedRoot resolves repository authorization before validating the repository spelling or touching pointer, root, count, index, or member bytes.

func OpenAuthorizedService 

func OpenAuthorizedService(
	ctx context.Context,
	root, repository, serviceKey string,
	authorize AuthorizeRepository,
) (Root, ServiceReceipt, *ServiceMember, error)

OpenAuthorizedService preserves the same authorization-first boundary and returns only the selected service partition after a final pointer fence.

func ReadEvidenceContent 

func ReadEvidenceContent(
	ctx context.Context,
	dataDir, repository string,
	evidence Evidence,
) ([]byte, error)

ReadEvidenceContent returns only the exact cited byte span from the immutable blob object. It refuses external alternates and verifies the complete source digest before slicing, so a path or mutable branch is never trusted as citation authority.

func Recover 

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

func RestoreArchive 

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

func ValidateRoot 

func ValidateRoot(value Root) error

ValidateRoot exposes the closed root validator to later authorized readers.

Types

type ArchiveReport 

type ArchiveReport struct {
	Publications int
	Files        int
	Bytes        int64
	Omitted      int
}

func CreateArchive 

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

CreateArchive preserves only fully validated current relationship roots and their exact resolver/RPC/Kafka component generations. Corrupt or in-flight derived state is visibly omitted and never blocks the precious DB export.

func VerifyArchive 

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

type Authority 

type Authority struct {
	Repository                  string                         `json:"repository"`
	CatalogGenerationDigest     string                         `json:"catalog_generation_digest"`
	CatalogDigest               string                         `json:"catalog_digest"`
	CatalogSourceGeneration     string                         `json:"catalog_source_generation"`
	ServiceStateSetDigest       string                         `json:"service_state_set_digest"`
	ServiceStateSummaryDigest   string                         `json:"service_state_summary_digest,omitempty"`
	ServiceStateControlRevision uint64                         `json:"service_state_control_revision,omitempty"`
	ObservationGenerationDigest string                         `json:"observation_generation_digest"`
	ObservationManifestDigest   string                         `json:"observation_manifest_digest"`
	ObservationSourceDigest     string                         `json:"observation_source_digest"`
	ResolverGenerationDigest    string                         `json:"resolver_generation_digest"`
	ResolverRootDigest          string                         `json:"resolver_root_digest"`
	RPCGenerationDigest         string                         `json:"rpc_generation_digest"`
	RPCRootDigest               string                         `json:"rpc_root_digest"`
	KafkaGenerationDigest       string                         `json:"kafka_generation_digest"`
	KafkaRootDigest             string                         `json:"kafka_root_digest"`
	PolicyDigest                string                         `json:"policy_digest"`
	Upstream                    *downstreamauthority.Authority `json:"upstream,omitempty"`
}

type AuthorizeRepository 

type AuthorizeRepository func(context.Context, string) (bool, error)

type BuildRequest 

type BuildRequest struct {
	Root     string
	Catalog  servicecatalog.Publication
	States   []servicecatalog.ServiceState
	Resolver *resolvernamespace.Publication
	RPC      *rpccallerposting.Publication
	Kafka    *kafkatopicposting.Publication
}

type BuildRequestV2 

type BuildRequestV2 struct {
	BuildRequest
	Upstream       downstreamauthority.Authority
	ServiceSummary servicecatalog.RepositoryState
	Prior          *Publication
}

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)

Acquire pins the exact current relationship root. Sparse service readers may share this bounded control open; selected members remain independently validated by Publication.OpenService.

func (*Cache) AcquireGeneration 

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

AcquireGeneration pins an exact immutable historical generation. The entry is installed before filesystem access so a lifecycle sweep that starts afterward observes the pin; if a sweep already renamed the generation, the open fails and the provisional entry is removed.

func (*Cache) BeginRetire 

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

BeginRetire atomically excludes a new historical lease after proving that no current or provisional lease exists. The caller holds the returned guard through rename, after which AcquireGeneration can no longer open the old immutable path even when the guard is released.

func (*Cache) Pinned 

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

type Evidence 

type Evidence struct {
	Kind                  string                 `json:"kind"`
	Plane                 string                 `json:"plane"`
	Class                 string                 `json:"class"`
	Reason                string                 `json:"reason,omitempty"`
	Path                  string                 `json:"path"`
	ObjectID              string                 `json:"object_id"`
	ContentDigest         string                 `json:"content_digest"`
	Span                  sourceobservation.Span `json:"span"`
	SourceRole            string                 `json:"source_role"`
	Operation             string                 `json:"operation,omitempty"`
	CandidateOperations   []string               `json:"candidate_operations,omitempty"`
	DeclarationPath       string                 `json:"declaration_path,omitempty"`
	DeclarationLineage    string                 `json:"declaration_lineage,omitempty"`
	ResolverRecordDigests []string               `json:"resolver_record_digests,omitempty"`
	TopicSpelling         string                 `json:"topic_spelling,omitempty"`
	GroupIDSpelling       string                 `json:"group_id_spelling,omitempty"`
	Library               string                 `json:"library,omitempty"`
	Shape                 string                 `json:"shape,omitempty"`
	Binding               string                 `json:"binding,omitempty"`
	PostingDigest         string                 `json:"posting_digest"`
}

Evidence is the common exact source record behind one relationship projection. Protocol-specific classifications remain explicit; no RPC declaration or Kafka broker/runtime identity is inferred by this adapter.

type EvidenceReader 

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

func (*EvidenceReader) ReadPage 

func (reader *EvidenceReader) ReadPage(
	ctx context.Context,
	projections []Projection,
) (map[string]Evidence, error)

ReadPage groups projection evidence by protocol/plane and lookup key. Each selected component member is read once even when many page rows share it.

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
	ReleasedPinOwner string
}

func SweepLifecycle 

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

SweepLifecycle removes at most one unrooted relationship or orphaned component generation. The caller holds the exclusive mutation lock. Relationship roots are renamed out of authority before bounded draining; component generations are deleted only after no remaining root references them and no resolver pointer/marker protects them.

type Marker 

type Marker struct {
	Schema  string  `json:"schema"`
	Pointer Pointer `json:"pointer"`
	Digest  string  `json:"digest"`
}

type MemberReceipt 

type MemberReceipt struct {
	Bucket        int    `json:"bucket"`
	Name          string `json:"name"`
	RecordCount   int    `json:"record_count"`
	ContentBytes  int64  `json:"content_bytes"`
	ContentDigest string `json:"content_digest"`
}

type PinChecker 

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

type Placement 

type Placement struct {
	Path    string         `json:"path"`
	Unowned bool           `json:"unowned"`
	Claims  []ServiceClaim `json:"claims"`
}

type Pointer 

type Pointer struct {
	Schema           string `json:"schema"`
	Repository       string `json:"repository"`
	GenerationDigest string `json:"generation_digest"`
	RootDigest       string `json:"root_digest"`
	RootName         string `json:"root_name"`
	Digest           string `json:"digest"`
}

type Policy 

type Policy struct {
	Schema                    string `json:"schema"`
	RepositoryBuckets         int    `json:"repository_buckets"`
	MaxServices               int    `json:"max_services"`
	MaxProjectionRecords      int    `json:"max_projection_records"`
	MaxServiceReferences      int    `json:"max_service_references"`
	MaxTotalServiceReferences int    `json:"max_total_service_references"`
	MaxClaimsPerPlacement     int    `json:"max_claims_per_placement"`
	MaxRolesPerClaim          int    `json:"max_roles_per_claim"`
	MaxTextBytes              int    `json:"max_text_bytes"`
	MaxProjectionBytes        int    `json:"max_projection_bytes"`
	MaxReferenceBytes         int    `json:"max_reference_bytes"`
	MaxRepositoryMemberBytes  int    `json:"max_repository_member_bytes"`
	MaxServiceMemberBytes     int    `json:"max_service_member_bytes"`
	MaxRootBytes              int    `json:"max_root_bytes"`
	MaxGenerationBytes        int64  `json:"max_generation_bytes"`
	MaxResidentChargeBytes    int64  `json:"max_resident_charge_bytes"`
	PlacementRule             string `json:"placement_rule"`
}

func FrozenPolicy 

func FrozenPolicy() Policy

type Prepared 

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

func Build 

func Build(ctx context.Context, request BuildRequest) (*Prepared, error)

func BuildV2 

func BuildV2(ctx context.Context, request BuildRequestV2) (*Prepared, error)

func (*Prepared) Publish 

func (prepared *Prepared) Publish(ctx context.Context) (*Publication, error)

func (*Prepared) Root 

func (prepared *Prepared) Root() Root

type Projection 

type Projection struct {
	Schema        string     `json:"schema"`
	Kind          string     `json:"kind"` // rpc | kafka
	PostingDigest string     `json:"posting_digest"`
	Class         string     `json:"class"`
	Plane         string     `json:"plane"`
	LookupKey     string     `json:"lookup_key,omitempty"`
	Source        Placement  `json:"source"`
	Target        *Placement `json:"target,omitempty"`
	Digest        string     `json:"digest"`
}

type Publication 

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

func OpenCurrent 

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

func OpenGeneration 

func OpenGeneration(
	ctx context.Context,
	root, repository, generation, rootDigestValue string,
) (*Publication, error)

OpenGeneration opens one exact immutable relationship generation without consulting the mutable current pointer. Callers that expose the result must hold a Cache lease for the same generation so lifecycle collection cannot retire its bytes while a cursor, comparison, or citation still references them.

func (*Publication) ConfirmCurrent 

func (publication *Publication) ConfirmCurrent() error

ConfirmCurrent is the final response fence for a current-root read.

func (*Publication) OpenEvidenceReader 

func (publication *Publication) OpenEvidenceReader(
	ctx context.Context,
	dataDir string,
) (*EvidenceReader, error)

OpenEvidenceReader validates each exact component root once for a bounded response binding. Page reads still validate selected members.

func (*Publication) OpenService 

func (publication *Publication) OpenService(
	ctx context.Context,
	serviceKey string,
) (ServiceReceipt, *ServiceMember, error)

func (*Publication) ReadEvidence 

func (publication *Publication) ReadEvidence(
	ctx context.Context,
	dataDir string,
	projection Projection,
) (Evidence, error)

ReadEvidence resolves a projection through the exact component generation bound into its relationship root. The lookup opens one component root and one keyed member; it never walks a repository corpus or sibling member.

func (*Publication) ReadProjection 

func (publication *Publication) ReadProjection(
	ctx context.Context,
	digest string,
) (Projection, error)

ReadProjection validates and returns exactly one repository projection. It deliberately does not consult current.json: the caller may be resolving a citation against a lease-pinned historical generation. Current response paths call ConfirmCurrent immediately before emission.

func (*Publication) ReadProjections 

func (publication *Publication) ReadProjections(
	ctx context.Context,
	digests []string,
) (map[string]Projection, error)

ReadProjections batches one page of projection lookups by bucket so a page never rereads the same repository member once per row.

func (*Publication) ReadService 

func (publication *Publication) ReadService(
	ctx context.Context,
	serviceKey string,
) (ServiceReceipt, *ServiceMember, error)

ReadService validates one selected service member without consulting the mutable pointer. It is for a caller holding an exact generation lease.

func (*Publication) Root 

func (publication *Publication) Root() Root

type RecoveryPinStore 

type RecoveryPinStore interface {
	PinPartitionedExtractionRun(context.Context, string, string) error
	ReconcilePartitionedExtractionOwners(context.Context, []string) error
}

type RecoveryReport 

type RecoveryReport struct {
	Repositories int
	Completed    int
	Unavailable  int
	Invalid      int
}

func RecoverAll 

func RecoverAll(ctx context.Context, dataDir string, pins RecoveryPinStore) (RecoveryReport, error)

RecoverAll completes exact component/root markers under the caller's exclusive startup mutation lock. Corrupt derived state is counted and left invisible; component-only crash residue is independently discovered.

type RepositoryMember 

type RepositoryMember struct {
	Schema          string       `json:"schema"`
	AuthorityDigest string       `json:"authority_digest"`
	Bucket          int          `json:"bucket"`
	Projections     []Projection `json:"projections"`
	Digest          string       `json:"digest"`
}

type RoleClaim 

type RoleClaim struct {
	Role   string `json:"role"`
	Origin string `json:"origin"`
}

type Root 

type Root struct {
	Schema                 string           `json:"schema"`
	Authority              Authority        `json:"authority"`
	AuthorityDigest        string           `json:"authority_digest"`
	Policy                 Policy           `json:"policy"`
	RepositoryComplete     bool             `json:"repository_complete"`
	AllServicesComplete    bool             `json:"all_services_complete"`
	RepositoryMembers      []MemberReceipt  `json:"repository_members"`
	Services               []ServiceReceipt `json:"services"`
	ProjectionCount        int              `json:"projection_count"`
	ServiceCount           int              `json:"service_count"`
	CompleteServiceCount   int              `json:"complete_service_count"`
	EmptyServiceCount      int              `json:"empty_service_count"`
	FailedServiceCount     int              `json:"failed_service_count"`
	ServiceReferenceCount  int              `json:"service_reference_count"`
	EncodedRepositoryBytes int64            `json:"encoded_repository_bytes"`
	EncodedServiceBytes    int64            `json:"encoded_service_bytes"`
	GenerationDigest       string           `json:"generation_digest"`
	Digest                 string           `json:"digest"`
}

type Runtime 

type Runtime struct {
	DataDir        string
	Store          RuntimeStore
	Cache          *observationpublication.Cache
	InventoryCache *observationpublication.InventoryCacheV2
	Domains        []downstreamauthority.DomainIdentity
	Admit          func(context.Context) error
	Acquire        func(context.Context) (func(), error)
	// contains filtered or unexported fields
}

func (*Runtime) Handle 

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

Handle builds all repository-shared components and publishes the one atomic relationship root only after rechecking every precious authority fence.

func (*Runtime) Reconcile 

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

Reconcile binds current observation, catalog, and resolver controls to one durable one-chunk build. No derived member or index is opened on the no-op path; a worker performs the admitted full joins.

type RuntimeStore 

type RuntimeStore interface {
	store.GenerationSchedulerStore
	GetGenerationScheduleFailure(
		context.Context, string, string, string,
	) (*store.GenerationScheduleFailure, error)
	store.ServiceCatalogPublicationStore
	store.ServiceStateStore
	store.ResolverCatalogPublicationStore
	store.PartitionedEvidenceStore
}

RuntimeStore is the exact durable control surface used by the relationship worker. It deliberately excludes evidence and historical-generation scans.

type ServiceClaim 

type ServiceClaim struct {
	ServiceKey  string      `json:"service_key"`
	Disposition string      `json:"disposition"`
	Roles       []RoleClaim `json:"roles"`
}

type ServiceMember 

type ServiceMember struct {
	Schema            string             `json:"schema"`
	AuthorityDigest   string             `json:"authority_digest"`
	ServiceKey        string             `json:"service_key"`
	Incarnation       uint64             `json:"incarnation"`
	ServiceGeneration string             `json:"service_generation"`
	References        []ServiceReference `json:"references"`
	Digest            string             `json:"digest"`
}

type ServiceReceipt 

type ServiceReceipt struct {
	ServiceKey        string `json:"service_key"`
	Incarnation       uint64 `json:"incarnation"`
	ServiceGeneration string `json:"service_generation"`
	State             string `json:"state"` // complete | empty | failed
	Reason            string `json:"reason,omitempty"`
	Name              string `json:"name,omitempty"`
	ReferenceCount    int    `json:"reference_count"`
	ContentBytes      int64  `json:"content_bytes"`
	ContentDigest     string `json:"content_digest,omitempty"`
}

type ServiceReference 

type ServiceReference struct {
	Schema           string   `json:"schema"`
	ProjectionDigest string   `json:"projection_digest"`
	PostingDigest    string   `json:"posting_digest"`
	Kind             string   `json:"kind"`
	Plane            string   `json:"plane"`
	LookupKey        string   `json:"lookup_key,omitempty"`
	Participation    []string `json:"participation"` // source | target
	Digest           string   `json:"digest"`
}

type Unavailable 

type Unavailable struct {
	Schema   string                        `json:"schema"`
	Upstream downstreamauthority.Authority `json:"upstream"`
	Reason   string                        `json:"reason"`
	Prior    *Pointer                      `json:"prior,omitempty"`
	Digest   string                        `json:"digest"`
}

func MarkUnavailable 

func MarkUnavailable(
	ctx context.Context, root, repository string,
	upstream downstreamauthority.Authority,
) (Unavailable, error)

MarkUnavailable installs an explicit source-free non-authority before removing current.json. OpenCurrent checks this marker first, so neither a crash between those mutations nor an older valid generation can become a silent fallback for failed or prerequisite-unavailable upstream work.

func ReadUnavailable 

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

ReadUnavailable returns the exact source-free unavailable marker selected by a product reader after repository authorization. The result is safe for projection into a receipt and contains no member, observation, extraction, repository, or Git content reads.

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.