snapshotoffload

type DeletePrecondition struct {
	ETag      string
	Size      int64
	UpdatedAt time.Time
}

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

type GCOptions struct {
	Store  RetentionStore
	Prefix string
	Policy RetentionPolicy
	// VerifyPublishersUseClaims must confirm that every publisher sharing
	// this prefix uses object claims and all legacy in-flight publishes have
	// drained. It runs before each pass; fail it before allowing a downgrade
	// or a legacy publisher to resume. A missing verifier disables GC.
	VerifyPublishersUseClaims func(context.Context) error
	// Now defaults to time.Now. Injected so tests can drive the age
	// windows deterministically rather than by sleeping.
	Now    func() time.Time
	Logger *slog.Logger
}

type GCResult struct {
	GroupsScanned       int
	ManifestsScanned    int
	ManifestsDeleted    []string
	PayloadsDeleted     []string
	MalformedManifests  []string
	PayloadPhaseSkipped bool
	// SkipReason explains PayloadPhaseSkipped in operator-facing
	// terms. Empty when the phase ran.
	SkipReason string
	// ManifestsClaimedConcurrently counts expired manifests whose
	// compare-and-delete lost to a concurrent publish rewriting the
	// same key (an idempotent republish of the same index/term).
	ManifestsClaimedConcurrently int
	// PayloadsAwaitingSweep counts payloads that are eligible for
	// reclamation but are still serving the §5 two-pass sweep delay.
	// A steady non-zero value on a busy prefix is normal.
	PayloadsAwaitingSweep int
	// PayloadsClaimedConcurrently counts payloads that were eligible
	// for reclamation but whose compare-and-delete lost to a
	// concurrent publish reusing them. A steady non-zero value is
	// healthy dedup traffic, not an error.
	PayloadsClaimedConcurrently int
}

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

type Manifest struct {
	SchemaVersion  int               `json:"schema_version"`
	CreatedAt      time.Time         `json:"created_at"`
	SourceCluster  string            `json:"source_cluster,omitempty"`
	GroupID        uint64            `json:"group_id"`
	SnapshotIndex  uint64            `json:"snapshot_index"`
	SnapshotTerm   uint64            `json:"snapshot_term"`
	ConfState      ManifestConfState `json:"conf_state"`
	Payload        PayloadDescriptor `json:"payload"`
	BinaryVersion  string            `json:"binary_version,omitempty"`
	ManifestKey    string            `json:"manifest_key"`
	ManifestSHA256 string            `json:"manifest_sha256,omitempty"`
}

type ManifestConfState struct {
	Voters         []uint64 `json:"voters,omitempty"`
	Learners       []uint64 `json:"learners,omitempty"`
	VotersOutgoing []uint64 `json:"voters_outgoing,omitempty"`
	LearnersNext   []uint64 `json:"learners_next,omitempty"`
	AutoLeave      bool     `json:"auto_leave,omitempty"`
}

type ObjectClaim interface {
	Release(ctx context.Context) error
}

type ObjectClaimStore interface {
	AcquireObjectClaim(ctx context.Context, key string) (ObjectClaim, error)
}

type ObjectInfo struct {
	Key       string
	Size      int64
	UpdatedAt time.Time
	// SHA256 is optional for metadata-only Head/Get paths; PutObject returns it
	// when the writer verified the committed content.
	SHA256               string
	ServerSideEncryption string
	SSEKMSKeyID          string
	// ETag is the store's opaque version token, when it supplies one.
	// Retention uses it as the compare-and-delete precondition.
	ETag string
}

type ObjectRef struct {
	Key       string
	Size      int64
	UpdatedAt time.Time
	// ETag is the store's opaque version token for the exact bytes
	// observed. It is the precondition DeleteObjectIfUnmodified uses
	// to make reclamation a compare-and-delete rather than a
	// check-then-delete. Empty when the store cannot supply one, in
	// which case the size + mtime pair is the fallback precondition.
	ETag string
}

type ObjectRefresher interface {
	RefreshObject(ctx context.Context, key string, body io.Reader, opts PutOptions) (ObjectInfo, error)
}

type ObjectStore interface {
	PutObject(ctx context.Context, key string, body io.Reader, opts PutOptions) (ObjectInfo, error)
	GetObject(ctx context.Context, key string) (io.ReadCloser, ObjectInfo, error)
	HeadObject(ctx context.Context, key string) (ObjectInfo, bool, error)
}

type OffloadGroup struct {
	GroupID uint64
	DataDir string
	// IsLeader is the cheap pre-check made before opening the snapshot.
	IsLeader func() bool
	// VerifyLeader is the authoritative check, re-run immediately before the
	// manifest is committed. See PublishOptions.VerifyLeader.
	VerifyLeader func(context.Context) error
}

type PayloadDescriptor struct {
	Key          string `json:"key"`
	Bytes        int64  `json:"bytes"`
	SHA256       string `json:"sha256"`
	SourceCRC32C uint32 `json:"source_crc32c"`
}

type PublishOptions struct {
	Store         PublishStore
	DataDir       string
	Prefix        string
	GroupID       uint64
	SourceCluster string
	BinaryVersion string
	CreatedAt     time.Time
	SpoolDir      string
	// VerifyLeader, when set, is re-checked immediately before the manifest is
	// committed. §4 requires leadership to hold at that instant, not merely when
	// the snapshot was opened: spooling a multi-gigabyte payload takes long
	// enough to lose an election. Failing here can leave an unreferenced
	// content-addressed payload, which GC reclaims, but never a committed
	// manifest naming a snapshot this node no longer had the right to publish.
	VerifyLeader func(context.Context) error
	// SkipIfNotNewerThan suppresses the publish when the persisted
	// snapshot's index is not greater than this value. Zero disables
	// the check.
	//
	// The comparison happens after the export is opened but BEFORE
	// the payload is spooled, which is the whole point: a scheduler
	// that ticks every 15 minutes over an unchanged snapshot would
	// otherwise re-read, re-hash and re-fsync a multi-gigabyte
	// payload every tick just to discover the object store already
	// has it.
	SkipIfNotNewerThan uint64
}

type PublishStore interface {
	ObjectStore
	ObjectClaimStore
	ObjectRefresher
}

type PutOptions struct {
	Size        int64
	SHA256      string
	ContentType string
}

type RestoreOptions struct {
	Store       ObjectStore
	ManifestKey string
	Manifest    *Manifest
	DataDir     string
	Peers       []etcdraftengine.Peer

	// ExpectGroupID is the Raft group the operator believes this data
	// directory belongs to. Required.
	//
	// Nothing downstream carries the group's identity: the restored
	// artifacts record index, term, peers and payload hash, but the group
	// comes only from the manifest, and startup derives the group from the
	// directory layout instead. So a group-2 manifest restored into a
	// group-1 data directory produces a perfectly valid-looking
	// directory that startup then loads as group 1 -- the wrong physical
	// FSM under another group's routing identity, with no error anywhere.
	// The only place that mistake can be caught is here, against what the
	// operator says they intended.
	//
	// A pointer because group 0 is a real group (the dedicated TSO group),
	// so zero cannot double as "unset".
	ExpectGroupID *uint64

	// ExpectSourceCluster is the cluster the operator believes this manifest
	// came from. Required.
	//
	// The group check alone is not enough when one bucket holds backups from
	// several clusters, even under different prefixes: another cluster's
	// manifest for the SAME group id passes it, and the restore produces a
	// structurally valid directory that startup then loads as this cluster's
	// FSM. Nothing downstream records the source cluster either, so this is
	// the only place the mistake is detectable.
	ExpectSourceCluster string

	// AllowLegacySourceClusterless permits restoring a legacy nonzero-group
	// manifest created before source-cluster became mandatory. The caller must
	// opt in explicitly because the manifest cannot prove its source cluster.
	AllowLegacySourceClusterless bool
}

type RetentionPolicy struct {
	MinGenerations int
	MaxAge         time.Duration
	PayloadGrace   time.Duration
	// MinMarkAge is the §5 two-pass sweep delay: a payload must have
	// been marked for at least this long, and must not have changed
	// since the mark, before it may be reclaimed.
	//
	// It exists because no conditional-delete primitive on a
	// general-purpose S3 bucket can detect a content-preserving
	// refresh: `If-Match` compares a content-derived ETag, which a
	// republish of identical bytes leaves untouched, and
	// `IfMatchLastModifiedTime` is directory-buckets only. Comparing
	// the object's observed state ACROSS passes is what detects that
	// refresh, so it must exceed the longest plausible publish.
	MinMarkAge time.Duration
}

type RetentionStore interface {
	ObjectStore
	ObjectClaimStore
	// ValidateRetention checks backend-specific reclamation prerequisites
	// before each pass. Read-only restore must not require these permissions.
	ValidateRetention(ctx context.Context) error
	ListObjects(ctx context.Context, prefix string) ([]ObjectRef, error)
	DeleteObject(ctx context.Context, key string) error

	// DeleteObjectIfUnmodified deletes key only if it still matches
	// cond, and returns ErrObjectModified otherwise.
	//
	// This is what makes payload reclamation safe against a
	// concurrent publish. The grace window alone is not enough: a
	// publisher that reuses a content-addressed payload refreshes the
	// object to restart its grace, and an unconditional delete can
	// still land between GC observing the old state and the publisher
	// committing its manifest — leaving a committed manifest pointing
	// at bytes that no longer exist. Making the delete conditional on
	// the exact state GC validated turns that race into a skip.
	DeleteObjectIfUnmodified(ctx context.Context, key string, cond DeletePrecondition) error
}

type S3ObjectClient interface {
	PutObject(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error)
	GetObject(context.Context, *s3.GetObjectInput, ...func(*s3.Options)) (*s3.GetObjectOutput, error)
	HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
	CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error)
	UploadPart(context.Context, *s3.UploadPartInput, ...func(*s3.Options)) (*s3.UploadPartOutput, error)
	CompleteMultipartUpload(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error)
	AbortMultipartUpload(context.Context, *s3.AbortMultipartUploadInput, ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error)
	ListObjectsV2(context.Context, *s3.ListObjectsV2Input, ...func(*s3.Options)) (*s3.ListObjectsV2Output, error)
	DeleteObject(context.Context, *s3.DeleteObjectInput, ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)
	GetBucketVersioning(context.Context, *s3.GetBucketVersioningInput, ...func(*s3.Options)) (*s3.GetBucketVersioningOutput, error)
}

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

type S3StoreConfig struct {
	Client                 S3ObjectClient
	Bucket                 string
	Region                 string
	Endpoint               string
	Profile                string
	ForcePathStyle         bool
	AccessKeyID            string
	SecretAccessKey        string
	SessionToken           string
	ServerSideEncryption   string
	SSEKMSKeyID            string
	DisableChecksumHeaders bool
	HTTPClientDecorator    func(aws.HTTPClient) (aws.HTTPClient, error)
	// AllowVersionedBucketWithLifecycle is an explicit operator assertion
	// that the bucket has a noncurrent-version expiration lifecycle. The
	// GC rejects Enabled/Suspended versioning by default because keyed
	// retention deletes otherwise create delete markers without reclaiming
	// bytes.
	AllowVersionedBucketWithLifecycle bool
}

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

type SchedulerObserver interface {
	ObserveSnapshotOffloadPublished(groupID, index uint64, payloadBytes int64, elapsed time.Duration)
	ObserveSnapshotOffloadSkipped(groupID uint64, reason string)
	ObserveSnapshotOffloadFailed(groupID uint64, err error)
}

type SchedulerOption func(*Scheduler)