cluster

type Claim struct {
	Namespace       string
	Key             []byte
	OwnerInstanceID string
	OwnerAddress    string
	Fence           uint64
}

type ClaimStore interface {
	// AcquireClaim atomically acquires (namespace, key) for self. It succeeds
	// if the claim is absent, released, or already held by self (an owner's
	// re-acquire returns the existing claim with the fence unchanged).
	//
	// If takeover is non-nil, the claim may additionally be displaced from
	// takeover.InstanceID, but only if — atomically at commit time — that
	// member's registry record is absent or its heartbeat counter still equals
	// takeover.HeartbeatCounter. The fence increments on every change of
	// owner, including a fresh acquire after a release.
	//
	// When the claim is (still) held by another member, the current holder is
	// returned alongside ErrClaimHeld.
	AcquireClaim(ctx context.Context, namespace string, key []byte, self *Member, takeover *TakeoverTarget) (*Claim, error)

	// GetClaim returns the active claim for (namespace, key), or
	// ErrClaimNotFound if the key is unclaimed or released.
	GetClaim(ctx context.Context, namespace string, key []byte) (*Claim, error)

	// ReleaseClaim releases the claim if it is currently held by instanceID; a
	// release by anyone else is a no-op. The fence survives release, so a
	// later acquire of the same key still increments monotonically.
	ReleaseClaim(ctx context.Context, namespace string, key []byte, instanceID string) error
}

type Member struct {
	InstanceID string
	Address    string
	Labels     map[string]string
	Draining   bool
}

type MemberRecord struct {
	Member
	HeartbeatCounter uint64
}

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

type MembershipConfig struct {
	// HeartbeatInterval is how often the member's own heartbeat counter is
	// advanced. Default 5s.
	HeartbeatInterval time.Duration

	// PollInterval is how often the registry is re-read to refresh the live
	// set. Must be well under LivenessWindow. Default 2s.
	PollInterval time.Duration

	// LivenessWindow is how long a peer's heartbeat counter may sit unchanged
	// (in this process's own observations) before the peer is considered dead.
	// Default 15s.
	LivenessWindow time.Duration

	// SelfUnhealthyAfter is how long this process's own heartbeat writes may
	// fail before it must assume the rest of the cluster considers it dead and
	// stop serving owned keys. Defaulted to the session-gap floor so Do starts
	// refusing owned keys no later than a suspicious peer could displace them;
	// raising it toward LivenessWindow trades a wider potential dual-serving
	// window for fewer fallback-path detours during store blips. Default:
	// min(SessionGapThreshold, LivenessWindow).
	SelfUnhealthyAfter time.Duration

	// SessionGapThreshold is the gap between successful heartbeat writes at
	// which the member treats its own liveness session as interrupted (firing
	// OnSessionLost, which sheds local ownership) — whether the intervening
	// beats were merely delayed or failing outright. Aligned by default with
	// the suspicion floor (HeartbeatInterval + PollInterval, capped at an
	// explicitly set SelfUnhealthyAfter): a heartbeat gap past that floor is
	// exactly the window in which a peer holding a failed-forward report may
	// have displaced this member's claims while it still looked healthy to
	// itself — shedding then bounds any such dual ownership at the gap's
	// length instead of IdleTTL. Default: HeartbeatInterval + PollInterval,
	// capped at SelfUnhealthyAfter when that is set.
	SessionGapThreshold time.Duration

	// MemberGCAfter is how long a member's heartbeat may sit unchanged before
	// any observer actively deletes its registry record. Store-level TTLs are
	// lazy (DynamoDB's can lag days), and a lingering corpse record misleads
	// freshly started observers — a first-sighted record is presumed live for
	// a full LivenessWindow. Must be much larger than LivenessWindow; a
	// wrongly deleted (merely wedged) member re-registers on its next
	// heartbeat. Default: 10 × LivenessWindow.
	MemberGCAfter time.Duration
}

type NamespaceHooks struct {
	OnAcquired func(ctx context.Context, key []byte)
	OnReleased func(ctx context.Context, key []byte)
}

type NotOwnerError struct {
	Redirect *Member
}

type Overrides interface {
	Get(ctx context.Context, namespace string, key []byte) (selector map[string]string, ok bool, err error)
}

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

type OwnershipConfig struct {
	// IdleTTL is how long an owned key may go unused before its claim is
	// released. Deliberately long: claims are sticky, and only liveness is
	// fresh — a short TTL converts the coordination plane's cheapest property
	// (claims touched only at activity boundaries) into churn. Default 15m.
	IdleTTL time.Duration

	// ReapInterval is how often idle claims are scanned for release.
	// Default 1m.
	ReapInterval time.Duration

	// DrainDeadline bounds how long a key's in-flight work may delay its
	// release during handoff or shutdown; past it the release proceeds
	// regardless (the consumer's store-serialized fallback makes a forced
	// release safe). It is also the timeout on the detached contexts handed
	// to OnReleased flushes and claim-release writes — a context-honoring
	// hook therefore bounds each release at roughly 3× DrainDeadline
	// (quiesce, then flush). Default 5s.
	DrainDeadline time.Duration

	// SuspicionWindow is the minimum heartbeat staleness at which a holder
	// reported unreachable (via NoteUnreachable) becomes takeover-eligible
	// ahead of the full LivenessWindow. Default 5s; floored at construction
	// to the membership's HeartbeatInterval + PollInterval — anything lower
	// leaves zero tolerance for ordinary heartbeat jitter, displacing owners
	// whose beat merely landed late.
	SuspicionWindow time.Duration

	// RedirectCacheTTL bounds how long a non-owner reuses a resolved redirect
	// for a key without re-reading the claim. Keeps the forward path from
	// paying a store read per request on hot keys; a stale redirect
	// self-corrects (the target answers NotOwner, or a failed forward
	// invalidates via NoteUnreachable). Default 1s.
	RedirectCacheTTL time.Duration
}

type Placement func(ctx context.Context, key []byte, candidates []*Member) ([]*Member, error)

type RegistryStore interface {
	// PutMember creates (or replaces) the member's record with the given
	// heartbeat counter. Instance IDs are unique per process incarnation, so a
	// replace only ever overwrites this process's own registration.
	//
	// The counter must never repeat within an instance ID's lifetime: takeover
	// evidence is "counter X was observed stale", checked as equality at
	// commit time, so a re-registration after record deletion must resume
	// strictly above the last value it wrote — resetting would let a peer's
	// stale prior-epoch evidence displace a live owner.
	PutMember(ctx context.Context, member *Member, heartbeatCounter uint64) error

	// Heartbeat advances the member's heartbeat counter and returns the new
	// value. Returns ErrMemberNotFound if the record no longer exists.
	Heartbeat(ctx context.Context, instanceID string) (uint64, error)

	// SetDraining updates the member's draining flag. A draining member is
	// excluded from routing candidacy but remains live, so claims it still
	// holds stay valid while it hands them off. Returns ErrMemberNotFound if
	// the record no longer exists.
	SetDraining(ctx context.Context, instanceID string, draining bool) error

	// DeleteMember removes the member's record. Idempotent.
	DeleteMember(ctx context.Context, instanceID string) error

	// GetMembers returns all registered member records.
	GetMembers(ctx context.Context) ([]*MemberRecord, error)
}

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

type Store interface {
	RegistryStore
	ClaimStore
	SubscriptionStore
}

type Subscription struct {
	Namespace  string
	Key        []byte
	InstanceID string
	// Address is the member's dialable endpoint, resolved by the Subscriptions
	// runtime from the membership view at resolution time — rows don't store
	// it. It is set on every Subscription a live resolution returns, and empty
	// on raw store reads.
	Address string
}

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

type SubscriptionStore interface {
	// PutSubscription registers (or reasserts) the instance's interest row for
	// the topic. Idempotent upsert. Rows carry identity only — consumers
	// resolve dial addresses through the membership view at resolution time.
	PutSubscription(ctx context.Context, namespace string, key []byte, instanceID string) error

	// PutSubscriptions registers the instance's interest rows for every listed
	// topic, as PutSubscription does for one, in as few store round trips as
	// the backend allows. Duplicate topics collapse. It is not atomic: on
	// error an arbitrary subset of the rows may have been written — acceptable
	// because a subscription row is interest, not ownership, and the runtime
	// sweeps its own refcount-less rows at resolution time.
	PutSubscriptions(ctx context.Context, topics []SubscriptionTopic, instanceID string) error

	// DeleteSubscription removes the member's interest row for the topic.
	// Idempotent.
	DeleteSubscription(ctx context.Context, namespace string, key []byte, instanceID string) error

	// DeleteSubscriptions removes the member's interest rows for every listed
	// topic, as DeleteSubscription does for one, in as few store round trips
	// as the backend allows. Duplicate topics collapse; idempotent. Not
	// atomic: on error an arbitrary subset of the rows may have been deleted
	// — the survivors stop counting once the member stops heartbeating, and
	// are swept as corpse rows.
	DeleteSubscriptions(ctx context.Context, topics []SubscriptionTopic, instanceID string) error

	// GetSubscribers returns every interest row for the topic, dead members'
	// rows included — liveness filtering is the caller's job.
	GetSubscribers(ctx context.Context, namespace string, key []byte) ([]*Subscription, error)
}

type SubscriptionTopic struct {
	Namespace string
	Key       []byte
}

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

type SubscriptionsConfig struct {
	// CacheTTL bounds how long a resolved subscriber set is reused without
	// re-reading the registry, so the publish path pays at most one registry
	// read per topic per TTL no matter the event rate. Empty results are
	// cached too — publishes toward offline users must not read per event.
	// The staleness is covered by the consumer's pull backstop: a just-opened
	// stream misses at most one TTL of events before its delta sync heals it.
	// Default 250ms — hot-topic read load at 4 reads/s per publisher is still
	// orders of magnitude under DynamoDB's per-partition ceiling, so the
	// freshness is nearly free; shrinking it much further buys latency the
	// delta sync already covers.
	CacheTTL time.Duration

	// RowGCAfter is how long a row's member may sit continuously outside this
	// observer's live view before the row is treated as a crashed instance's
	// leftover and deleted at resolution time. Graceful drains remove their
	// own rows; this sweeps what crashes leave behind on topics still being
	// published to. Floored at construction well above the liveness window so
	// a merely-slow member is never swept — and a wrongly swept row is
	// re-asserted by its owner on its next liveness session recovery. Default:
	// 10 × the membership's LivenessWindow, resolved at construction.
	RowGCAfter time.Duration
}

type TakeoverTarget struct {
	InstanceID       string
	HeartbeatCounter uint64
}