Documentation
¶
Overview ¶
Package sessionstore persists the fenced DeviceSession state machine.
Index ¶
- Variables
- type AcquireInput
- type ExecutionFailure
- type FrameContent
- type FrameContentRequest
- type FrameData
- type FrameWork
- type Identity
- type IdentityAdmin
- type InputWork
- type MutationInput
- type NodeLease
- type Postgres
- func (store *Postgres) Acquire(ctx context.Context, input AcquireInput) (Result, error)
- func (store *Postgres) ActivateNode(ctx context.Context, lease NodeLease) error
- func (store *Postgres) Apply(ctx context.Context, input MutationInput) (Result, error)
- func (store *Postgres) ApplySchema(ctx context.Context) error
- func (store *Postgres) ClaimFrame(ctx context.Context, lease NodeLease, claimFor time.Duration) (FrameWork, error)
- func (store *Postgres) ClaimInput(ctx context.Context, lease NodeLease, claimFor time.Duration) (InputWork, error)
- func (store *Postgres) Close()
- func (store *Postgres) CompleteFrame(ctx context.Context, work FrameWork, frame FrameData) error
- func (store *Postgres) CompleteInput(ctx context.Context, work InputWork, result string, latency time.Duration, ...) error
- func (store *Postgres) CurrentTime(ctx context.Context) (time.Time, error)
- func (store *Postgres) DeactivateNode(ctx context.Context, lease NodeLease) error
- func (store *Postgres) Events(ctx context.Context, tenantID, principalID, sessionID string, after int64) ([]devicesessionv1.Event, error)
- func (store *Postgres) ExpireSessions(ctx context.Context, limit int) (int64, error)
- func (store *Postgres) FailFrame(ctx context.Context, work FrameWork, code string, retryable bool, ...) error
- func (store *Postgres) FrameContent(ctx context.Context, input FrameContentRequest) (FrameContent, error)
- func (store *Postgres) HeartbeatNode(ctx context.Context, lease NodeLease, leaseFor time.Duration) error
- func (store *Postgres) ListIdentities(ctx context.Context) ([]Identity, error)
- func (store *Postgres) MarkDisconnected(ctx context.Context, ...) error
- func (store *Postgres) Ping(ctx context.Context) error
- func (store *Postgres) RecoverAmbiguousInputs(ctx context.Context, lease NodeLease, olderThan time.Duration) (int64, error)
- func (store *Postgres) RegisterDevice(ctx context.Context, tenantID, resourceID string, lease NodeLease, ...) error
- func (store *Postgres) RegisterNode(ctx context.Context, nodeID, address string, leaseFor time.Duration) (NodeLease, error)
- func (store *Postgres) RequireReadyNode(ctx context.Context, lease NodeLease) error
- func (store *Postgres) ReserveTokenNonce(ctx context.Context, fingerprint, nonce string, ttl time.Duration) (TokenWindow, error)
- func (store *Postgres) ResolveIdentity(ctx context.Context, fingerprint string) (Identity, error)
- func (store *Postgres) RevokeIdentity(ctx context.Context, fingerprint string, at time.Time) error
- func (store *Postgres) StartInput(ctx context.Context, work InputWork) error
- func (store *Postgres) UpsertIdentity(ctx context.Context, identity Identity) error
- func (store *Postgres) ValidateSessionAccess(ctx context.Context, ...) error
- func (store *Postgres) WaitEvents(ctx context.Context, tenantID, principalID, sessionID string, after int64, ...) ([]devicesessionv1.Event, bool, error)
- func (store *Postgres) WaitForExecutionQuiescence(ctx context.Context, lease NodeLease, olderThan, poll time.Duration) error
- func (store *Postgres) WaitForWork(ctx context.Context, nodeID string, wait time.Duration) error
- func (store *Postgres) WaitInputActive(ctx context.Context, work InputWork, poll time.Duration) (bool, error)
- type Result
- type Session
- type Store
- type TokenWindow
Constants ¶
This section is empty.
Variables ¶
View Source
var ( ErrNotFound = errors.New("resource not found") ErrBusy = errors.New("resource already leased") ErrFenced = errors.New("stale lease generation or fence") ErrConflict = errors.New("idempotency key conflict") ErrExpired = errors.New("binding or lease expired") ErrInvalidState = errors.New("invalid session state transition") ErrIdentityRevoked = errors.New("identity mapping is revoked") ErrInvalidArgument = errors.New("invalid request argument") ErrBackpressure = errors.New("runtime backpressure limit reached") )
Functions ¶
This section is empty.
Types ¶
type AcquireInput ¶
type AcquireInput struct {
TenantID string
PrincipalID string
AuthProfileID string
ChannelBindingSHA256 string
RequestNonce string
BindingExpiresAt time.Time
ResourceID string
RequestedCapabilities []string
IdempotencyKey string
ReleaseIdempotencyKey string
LeaseDuration time.Duration
HeartbeatInterval time.Duration
}
type ExecutionFailure ¶
type FrameContent ¶
type FrameContentRequest ¶
type IdentityAdmin ¶
type InputWork ¶
type InputWork struct {
SessionID string
TenantID string
ResourceID string
RequestSequence int64
RequestID string
IdempotencyKey string
Generation int64
FencingTokenSHA256 string
StreamEpoch int64
FrameSequence int64
Command string
CommandPayload json.RawMessage
ClaimedBy string
WorkerEpoch int64
ClaimGeneration int64
StartedAt time.Time
}
type MutationInput ¶
type MutationInput struct {
SessionID string
TenantID string
PrincipalID string
ChannelBindingSHA256 string
RequestNonce string
BindingExpiresAt time.Time
RequestID string
Type string
IdempotencyKey string
Generation int64
FencingTokenSHA256 string
Payload json.RawMessage
CommandPayload json.RawMessage
RequestedExtension time.Duration
LastAcknowledgedEvent int64
}
type NodeLease ¶
NodeLease identifies one live worker process. WorkerEpoch changes whenever an expired node identity is taken over, fencing every operation issued by an older process that used the same NodeID.
type Postgres ¶
func (*Postgres) ActivateNode ¶
func (*Postgres) ClaimFrame ¶
func (*Postgres) ClaimInput ¶
func (*Postgres) CompleteFrame ¶
func (*Postgres) CompleteInput ¶
func (*Postgres) CurrentTime ¶
func (*Postgres) DeactivateNode ¶
func (*Postgres) ExpireSessions ¶
func (*Postgres) FrameContent ¶
func (store *Postgres) FrameContent(ctx context.Context, input FrameContentRequest) (FrameContent, error)
func (*Postgres) HeartbeatNode ¶
func (*Postgres) ListIdentities ¶
func (*Postgres) MarkDisconnected ¶
func (store *Postgres) MarkDisconnected(ctx context.Context, tenantID, principalID, sessionID, channelBinding, requestNonce string, bindingExpiresAt time.Time, generation int64, fence, reason string) error
MarkDisconnected records transport loss without releasing or changing the fence. Repeated observations are idempotent; reconnect remains client-driven.
func (*Postgres) RecoverAmbiguousInputs ¶
func (*Postgres) RegisterDevice ¶
func (*Postgres) RegisterNode ¶
func (*Postgres) RequireReadyNode ¶
func (*Postgres) ReserveTokenNonce ¶
func (*Postgres) ResolveIdentity ¶
func (*Postgres) RevokeIdentity ¶
func (*Postgres) StartInput ¶
StartInput is the durable no-return point. A stale claimed job may be retried; a stale executing job is never re-executed because the external mutation may already have happened.
func (*Postgres) UpsertIdentity ¶
func (*Postgres) ValidateSessionAccess ¶
func (*Postgres) WaitEvents ¶
func (*Postgres) WaitForExecutionQuiescence ¶
func (*Postgres) WaitForWork ¶
type Result ¶
type Result struct {
Session Session `json:"session"`
Event devicesessionv1.Event `json:"event"`
Replay bool `json:"replay"`
Queued bool `json:"queued,omitempty"`
}
type Session ¶
type Session struct {
SessionID string `json:"session_id"`
TenantID string `json:"tenant_id"`
PrincipalID string `json:"principal_id"`
AuthProfileID string `json:"auth_profile_id"`
ChannelBindingSHA256 string `json:"channel_binding_sha256"`
RequestNonce string `json:"request_nonce"`
BindingExpiresAt time.Time `json:"binding_expires_at"`
ResourceID string `json:"resource_id"`
OwnerNodeID string `json:"owner_node_id,omitempty"`
OwnerWorkerEpoch int64 `json:"owner_worker_epoch,omitempty"`
LeaseID string `json:"lease_id"`
Generation int64 `json:"generation"`
FencingTokenSHA256 string `json:"fencing_token_sha256"`
ReleaseIdempotencyKey string `json:"release_idempotency_key"`
Capabilities []string `json:"capabilities"`
Status string `json:"status"`
StreamEpoch int64 `json:"stream_epoch"`
AcquiredAt time.Time `json:"acquired_at"`
LeaseExpiresAt time.Time `json:"lease_expires_at"`
HeartbeatInterval time.Duration `json:"heartbeat_interval"`
}
type Store ¶
type Store interface {
Ping(context.Context) error
CurrentTime(context.Context) (time.Time, error)
ReserveTokenNonce(context.Context, string, string, time.Duration) (TokenWindow, error)
Acquire(context.Context, AcquireInput) (Result, error)
Apply(context.Context, MutationInput) (Result, error)
Events(context.Context, string, string, string, int64) ([]devicesessionv1.Event, error)
ResolveIdentity(context.Context, string) (Identity, error)
ValidateSessionAccess(context.Context, string, string, string, string, string, time.Time, int64, string) error
}
Click to show internal directories.
Click to hide internal directories.