adapter

package
v0.0.0-...-6a70167 Latest Latest
Warning

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

 Go to latest Published: Aug 5, 2026 License: AGPL-3.0 Imports: 76 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	SQSPartitionActionSend    = "send"
	SQSPartitionActionReceive = "receive"
	SQSPartitionActionDelete  = "delete"

	SQSThrottleActionSend    = "send"
	SQSThrottleActionReceive = "receive"
	SQSThrottleActionDefault = "default"
)

SQS metric action labels mirror the values from monitoring/sqs.go. Re-declared so adapter call sites do not need a monitoring import; the observer interface validates the value at runtime so a drift between these constants and the monitoring side surfaces as a dropped observation rather than a wedge.

View Source 
const (
	// SqsQueueMetaPrefix prefixes queue-metadata records.
	SqsQueueMetaPrefix = "!sqs|queue|meta|"
	// SqsQueueGenPrefix prefixes the per-queue monotonic generation counter.
	// Bumped on DeleteQueue / PurgeQueue so keys from an older incarnation of
	// the same queue name cannot leak into a newly created queue.
	SqsQueueGenPrefix = "!sqs|queue|gen|"
	// SqsQueueSeqPrefix prefixes the per-queue FIFO sequence counter. Bumped
	// on every FIFO send and embedded in the message record so consumers can
	// reconstruct the producer's strict total order.
	SqsQueueSeqPrefix = "!sqs|queue|seq|"
	// SqsMsgDedupPrefix prefixes FIFO deduplication records. Each entry
	// stores the original message id and the dedup-window expiry; the
	// receive path is unaware of these — they only gate sends.
	SqsMsgDedupPrefix = "!sqs|msg|dedup|"
	// SqsMsgGroupPrefix prefixes the FIFO group-lock records. The lock is
	// held by at most one message per group, persists across visibility
	// expiries, and is only released on DeleteMessage / DLQ redrive /
	// retention expiry.
	SqsMsgGroupPrefix = "!sqs|msg|group|"
	// SqsMsgByAgePrefix prefixes the send-age index. Each entry is
	// keyed by (queue, gen, send_timestamp, message_id) so the reaper
	// can find every record whose retention deadline has elapsed with
	// one bounded scan, without having to load every message body.
	SqsMsgByAgePrefix = "!sqs|msg|byage|"
	// SqsQueueTombstonePrefix prefixes a generation-orphan marker.
	// DeleteQueue and PurgeQueue each write one (queue, gen) tombstone
	// in the same OCC transaction that supersedes that generation —
	// DeleteQueue tombstones the gen it removes the meta row at, and
	// PurgeQueue tombstones the pre-bump gen so the reaper can find
	// pre-purge orphans even if the queue is deleted before the next
	// reaper tick. The reaper enumerates these markers to clean up
	// orphan data / vis / byage / dedup / group keys for superseded
	// generations. The tombstone is itself deleted once the reaper
	// confirms no message-keyspace state remains for that (queue, gen).
	SqsQueueTombstonePrefix = "!sqs|queue|tombstone|"
)

SQS keyspace prefixes. Kept in sync with the naming in docs/design/2026_04_24_implemented_sqs_compatible_adapter.md.

View Source 
const (
	SqsPartitionedMsgDataPrefix  = "!sqs|msg|data|" + sqsPartitionedDiscriminator
	SqsPartitionedMsgVisPrefix   = "!sqs|msg|vis|" + sqsPartitionedDiscriminator
	SqsPartitionedMsgDedupPrefix = "!sqs|msg|dedup|" + sqsPartitionedDiscriminator
	SqsPartitionedMsgGroupPrefix = "!sqs|msg|group|" + sqsPartitionedDiscriminator
	SqsPartitionedMsgByAgePrefix = "!sqs|msg|byage|" + sqsPartitionedDiscriminator
)

SqsPartitionedMsg*Prefix mirrors each legacy SqsMsg*Prefix with the partitioned-keyspace discriminator inserted. Defined as full string constants (rather than runtime concatenation in each constructor) so the byte-layout invariant is asserted by the type system: a future rename of the discriminator must touch the constants here, not 6+ scattered string concatenations.

View Source 
const (
	SqsMsgDataPrefix = "!sqs|msg|data|"
	SqsMsgVisPrefix  = "!sqs|msg|vis|"
)

Message-keyspace prefixes. The data record holds the message body and state; the visibility index is a separate, visible_at-sorted key family so ReceiveMessage can find the next visible message with a single bounded prefix scan.

View Source 
const DefaultLuaPoolMaxIdle = 64

DefaultLuaPoolMaxIdle is the default upper bound on idle pooled *lua.LState instances retained for reuse. Each pooled state holds the base stdlib + redis/cjson/cmsgpack closures + per-state snapshot tables (globals / tables / metatables); empirically ~200 KiB of long-lived heap per state. 64 is sized to comfortably cover typical Redis-side EVAL/EVALSHA concurrency (one in-flight script per connection up to redcon's default worker pool) without retaining a long tail of warm states after a burst subsides.

Operators expecting sustained higher concurrency can raise the cap with --redisLuaMaxIdleStates; concurrency that exceeds the cap still works correctly — excess get() calls fall through to a fresh allocation (miss) and excess put() calls drop the state for the GC (drop). The cap therefore controls memory floor, not throughput ceiling.

Exported so main.go can use it as the default for the --redisLuaMaxIdleStates flag instead of duplicating the literal (which trips the mnd lint and creates a drift source if the adapter-side default changes).

View Source 
const DistributionFilePinnedHotspotSplitBoundary = "split_boundary"
View Source 
const (
	// S3BlobOffloadCapabilityName is the Admin GetClusterOverview capability
	// key used by mixed-version PUT admission before writing chunkref metadata.
	S3BlobOffloadCapabilityName = "feature_s3_blob_offload"
)

Variables

View Source 
var (
	// ErrAdminBucketAlreadyExists signals that AdminCreateBucket
	// targeted a name already in use. Maps to 409 Conflict.
	ErrAdminBucketAlreadyExists = errors.New("s3 admin: bucket already exists")
	// ErrAdminBucketNotFound signals that AdminDeleteBucket /
	// AdminPutBucketAcl targeted a missing bucket. Maps to 404.
	ErrAdminBucketNotFound = errors.New("s3 admin: bucket not found")
	// ErrAdminBucketNotEmpty signals that AdminDeleteBucket targeted
	// a bucket that still has objects. Maps to 409 Conflict to match
	// the SigV4 path's BucketNotEmpty response (the dashboard cannot
	// force a recursive delete; the operator must clean up first).
	ErrAdminBucketNotEmpty = errors.New("s3 admin: bucket is not empty")
	// ErrAdminInvalidBucketName signals that AdminCreateBucket got
	// a name that does not satisfy validateS3BucketName. Maps to 400.
	ErrAdminInvalidBucketName = errors.New("s3 admin: invalid bucket name")
	// ErrAdminInvalidACL signals that the ACL string did not pass
	// validateS3CannedAcl. Maps to 400 (the SigV4 path returns 501
	// NotImplemented for unsupported canned ACLs, but the admin API
	// is documented as private/public-read only and rejecting other
	// values as invalid input is a more useful contract for the
	// dashboard).
	ErrAdminInvalidACL = errors.New("s3 admin: invalid ACL")
)

Sentinel errors the admin write methods return so the bridge in main_admin.go can translate them into admin-package vocabulary without sniffing strings. Named separately from ErrAdminTableAlreadyExists / ErrAdminTableNotFound on the Dynamo side so a future per-resource role / status divergence does not require renaming both packages' callers.

View Source 
var ErrAdminDynamoNotFound = errors.New("dynamodb admin: target not found")

ErrAdminDynamoNotFound is the structured "item or table not found" sentinel admin handlers match against to render 404. The describe path uses (nil, false, nil) instead; this sentinel is for write-path failures where the missing target is observed only after the retry / commit attempt.

View Source 
var ErrAdminDynamoValidation = errors.New("dynamodb admin: invalid request")

ErrAdminDynamoValidation is returned when an admin entrypoint receives a structurally-bad request (missing key, malformed attribute, blank table name).

View Source 
var ErrAdminForbidden = errors.New("dynamodb admin: principal lacks required role")

ErrAdminForbidden is returned when the principal lacks the role required for the operation. Admin handlers translate this to 403 "forbidden" without leaking which field of the principal failed the check.

View Source 
var ErrAdminInvalidContinuationToken = errors.New("s3 admin: invalid continuation token")

ErrAdminInvalidContinuationToken signals AdminListObjects got a continuation token that is either malformed, references a different bucket / generation / prefix / delimiter than the current request, or names a readTS that has been MVCC-GC'd (store.ErrReadTSCompacted on the underlying scan). All three shapes map to 400 at the HTTP bridge; the wrapped message distinguishes the cases for operator diagnostics (claude-bot r1 + gemini r1 mediums on PR #811: the earlier ErrAdminInvalidBucketName reuse conflated token errors with bucket-name errors).

View Source 
var ErrAdminInvalidObjectKey = errors.New("s3 admin: invalid object key")

ErrAdminInvalidObjectKey signals an empty or otherwise malformed key. The HTTP bridge already rejects path-segment violations before reaching this layer; the sentinel exists so a direct Go caller (tests, future bridges) gets the documented contract.

View Source 
var ErrAdminNotLeader = errors.New("dynamodb admin: this node is not the raft leader")

ErrAdminNotLeader is returned by every write entrypoint when this node is not the verified Raft leader. The admin HTTP handler translates this to 503 + Retry-After: 1 today; the future AdminForward RPC catches it as the trigger to forward to the leader instead.

View Source 
var ErrAdminObjectNotFound = errors.New("s3 admin: object not found")

ErrAdminObjectNotFound signals that AdminGetObject / AdminDeleteObject targeted a missing object. AdminGetObject returns it as the error path; AdminDeleteObject treats absence as success (S3 delete is idempotent — matches AWS semantics and the SigV4 deleteObject path) so the sentinel does not surface on the delete return.

View Source 
var ErrAdminSQSNotFound = errors.New("sqs admin: queue not found")

ErrAdminSQSNotFound is returned by write entrypoints when the target queue does not exist. Maps to 404. The describe path uses the (nil, false, nil) tuple instead of this sentinel for the not-found signal, mirroring AdminDescribeTable.

View Source 
var ErrAdminSQSPurgeInProgress = errors.New("sqs admin: purge in progress")

ErrAdminSQSPurgeInProgress is the sentinel admin handlers match against for the 60-second PurgeQueue rate limit. errors.Is(err, ErrAdminSQSPurgeInProgress) returns true for any *PurgeInProgressError via the latter's Is method, so handlers can branch on the sentinel while extracting the typed RetryAfter duration from the wrapped value.

View Source 
var ErrAdminSQSValidation = errors.New("sqs admin: invalid queue name")

ErrAdminSQSValidation is returned when an admin entrypoint receives a request with a missing or syntactically-bad queue name. Maps to 400 in the admin HTTP handler.

View Source 
var ErrAdminTokenRequired = errors.New("admin token file required; pass --adminInsecureNoAuth to run without")

ErrAdminTokenRequired is returned by NewAdminServer helpers when the operator failed to supply a token and also did not opt into insecure mode.

View Source 
var ErrAdminUploadTooLarge = errors.New("s3 admin: object exceeds upload cap")

ErrAdminUploadTooLarge signals AdminPutObject received a body larger than adminS3UploadMaxBytes. The HTTP bridge maps this to 413 Payload Too Large.

View Source 
var ErrCollectionTooLarge = errors.New("collection too large")

ErrCollectionTooLarge is returned when a collection exceeds maxWideColumnItems.

View Source 
var ErrDeltaScanTruncated = errors.New("delta scan truncated: compaction required")

ErrDeltaScanTruncated is returned when the delta scan result is truncated, indicating that synchronous compaction is required before the operation can proceed.

View Source 
var ErrLeaderNotFound = errors.New("leader not found")
View Source 
var ErrNotLeader = errors.New("not leader")
View Source 
var ErrTxnTimestampOverflow = errors.New("txn timestamp overflow")

Functions

func AdminErrorMessage 

func AdminErrorMessage(err error) string

AdminErrorMessage extracts the human-readable message from a dynamoAPIError for surfacing back to the SPA. Returns "" when err is not a structured adapter error so callers fall back to a generic message instead of leaking arbitrary err.Error() output.

func AdminTokenAuth 

func AdminTokenAuth(token string) (grpc.UnaryServerInterceptor, grpc.StreamServerInterceptor)

AdminTokenAuth builds a gRPC unary+stream interceptor pair enforcing "authorization: Bearer <token>" metadata against the supplied token. An empty token disables enforcement; callers should pair that mode with a --adminInsecureNoAuth flag so operators knowingly opt in.

func AdvertisesHTFIFO 

func AdvertisesHTFIFO() bool

AdvertisesHTFIFO reports whether this binary's /sqs_health endpoint lists the htfifo capability. Mirror of the package- internal htfifoCapabilityAdvertised constant, exposed for the SQS leadership-refusal hook in main.go that uses this signal to decide whether to refuse leadership of any Raft group hosting a partitioned FIFO queue.

Stays a function (not an exported constant) so a future runtime override (env var, --no-htfifo flag for graceful degradation) can be threaded through here without changing the call site.

func IsAdminTableAlreadyExists 

func IsAdminTableAlreadyExists(err error) bool

IsAdminTableAlreadyExists reports whether err is the adapter's "table already exists" failure (ResourceInUseException). The bridge in main_admin.go uses this to map the adapter's internal error vocabulary onto admin's HTTP-facing sentinels without importing the package-private dynamoAPIError type.

func IsAdminTableNotFound 

func IsAdminTableNotFound(err error) bool

IsAdminTableNotFound is the ResourceNotFoundException counterpart for AdminDeleteTable / AdminDescribeTable mapped through the adapter's structured error chain.

func IsAdminValidation 

func IsAdminValidation(err error) bool

IsAdminValidation reports whether err is a validation failure the adapter signalled via ValidationException. Admin handlers map this to 400 + a sanitised message.

func S3BlobMinReplicasFromEnv 

func S3BlobMinReplicasFromEnv() (int, error)

S3BlobMinReplicasFromEnv returns zero when the dynamic Raft-quorum default should be used. Explicit values below two are rejected because a leader-only chunkblob is weaker than the legacy Raft data path.

func S3BlobOffloadEnabledFromEnv 

func S3BlobOffloadEnabledFromEnv() bool

S3BlobOffloadEnabledFromEnv returns the local rollout flag used by both the S3 write path and the Admin capability advertisement.

func S3BlobOffloadLocalCapability 

func S3BlobOffloadLocalCapability() bool

S3BlobOffloadLocalCapability reports whether this binary can serve the M1 offloaded PUT/GET path, including durable peer push and proxy-on-miss reads.

func S3BlobPeerTokenAuth 

func S3BlobPeerTokenAuth(token string) (grpc.UnaryServerInterceptor, grpc.StreamServerInterceptor)

S3BlobPeerTokenAuth protects the peer-only blob transport with a credential distinct from the read-only Admin token.

Types

type AdminAttribute 

type AdminAttribute struct {
	Name string
	Type string
}

AdminAttribute names a single primary-key or GSI key column. Type must be one of "S", "N", "B" — DynamoDB does not allow boolean or list keys and the adapter's existing schema validation enforces the same restriction at the next layer.

type AdminAttributeValue 

type AdminAttributeValue struct {
	S    *string                        `json:"S,omitempty"`
	N    *string                        `json:"N,omitempty"`
	B    []byte                         `json:"B,omitempty"`
	BOOL *bool                          `json:"BOOL,omitempty"`
	NULL *bool                          `json:"NULL,omitempty"`
	SS   []string                       `json:"SS,omitempty"`
	NS   []string                       `json:"NS,omitempty"`
	BS   [][]byte                       `json:"BS,omitempty"`
	L    []AdminAttributeValue          `json:"L,omitempty"`
	M    map[string]AdminAttributeValue `json:"M,omitempty"`
}

AdminAttributeValue mirrors the AWS DynamoDB-wire MessageAttribute shape. Each instance carries exactly one populated field for scalar types (S/N/B/BOOL/NULL), exactly one populated set (SS/NS/BS), or a recursive container (L, M). Wire-compatible with every AWS SDK and the existing SigV4 path's internal type.

Marshal / Unmarshal delegate to the internal attributeValue's AWS-wire-compatible codec (see MarshalJSON / UnmarshalJSON below) so the JSON struct tags here are decorative; they're kept as documentation of the wire shape.

func (AdminAttributeValue) MarshalJSON 

func (a AdminAttributeValue) MarshalJSON() ([]byte, error)

MarshalJSON / UnmarshalJSON delegate to the internal attributeValue's AWS-wire codec via the conversion functions. This is the only way to:

  1. preserve empty-but-present L/M ({"L": []} / {"M": {}}) on the wire — encoding/json's `omitempty` tag drops slices/maps of len 0 regardless of nil-ness (claude-bot r4 low on PR #805);
  2. enforce the AWS-wire "exactly one field set per AttributeValue" invariant — the internal MarshalJSON returns `invalid attribute value` for zero/multi-field inputs, which catches malformed Phase 3 HTTP handler output at the boundary rather than letting the SDK silently misinterpret a fall-back `"L": null` tag.

Caller audit: the only consumer of AdminAttributeValue's wire representation is the (not-yet-existing) Phase 3 admin HTTP bridge. The new fail-closed marshal error path is a Phase 3 boundary contract — Phase 2a's Go-level callers never marshal.

func (*AdminAttributeValue) UnmarshalJSON 

func (a *AdminAttributeValue) UnmarshalJSON(data []byte) error

type AdminBucketSummary 

type AdminBucketSummary struct {
	Name         string
	ACL          string
	CreatedAtHLC uint64
	Generation   uint64
	Region       string
	Owner        string
}

AdminBucketSummary is the bucket-level information the admin dashboard surfaces. It deliberately projects only the fields the dashboard needs so the package's wire-format types (s3BucketMeta, s3ListBucketsResult) stay internal.

CreatedAtHLC is the same physical-time-bearing HLC the bucket metadata persists; the admin HTTP handler formats it for the SPA. ACL is the canned-ACL string ("private" / "public-read") — the admin layer does not expand it into the AWS ACL XML grant tree because the dashboard renders the canned form directly.

type AdminCreateGSI 

type AdminCreateGSI struct {
	Name             string
	PartitionKey     AdminAttribute
	SortKey          *AdminAttribute
	ProjectionType   string
	NonKeyAttributes []string
}

AdminCreateGSI describes one global secondary index in an admin CreateTable request. SortKey is optional (hash-only GSI). When ProjectionType is "INCLUDE", NonKeyAttributes lists the projected attribute names; otherwise NonKeyAttributes is ignored.

type AdminCreateTableInput 

type AdminCreateTableInput struct {
	TableName    string
	PartitionKey AdminAttribute
	SortKey      *AdminAttribute
	GSI          []AdminCreateGSI
}

AdminCreateTableInput is the admin-facing CreateTable shape. The HTTP handler maps the design 4.2 JSON body into this struct, then AdminCreateTable converts it to the adapter's internal createTableInput. We do not pass the SigV4-flavoured wire struct directly because that struct's field names track AWS exactly and would be awkward for the admin SPA to author.

type AdminDescribeQueueOptions 

type AdminDescribeQueueOptions struct {
	// IncludeDLQSources turns on the paginated reverse-scan over
	// the queue-meta catalog that populates IsDLQ and DLQSources.
	// Cost is ceil(N/sqsQueueScanPageLimit) round-trips plus a
	// JSON decode per record (same envelope as scanQueueNamesAt),
	// so it is opt-in to keep AdminDescribeQueue calls from
	// callers that don't surface the DLQ relationship (e.g.
	// today's admin SPA, which drops the fields in the bridge)
	// O(1). Phase 4 (the SPA wiring) flips this on; reviewers
	// flagged the unconditional version as dead work otherwise
	// (Codex r1 P2, Gemini r1).
	IncludeDLQSources bool
}

AdminDescribeQueueOptions opts the Describe call into the more expensive lookups. Each option defaults to off so the cheap path (the only one Phase 2's bridge actually consumes) stays O(1).

type AdminGSISummary 

type AdminGSISummary struct {
	Name           string
	PartitionKey   string
	SortKey        string
	ProjectionType string
}

AdminGSISummary mirrors AdminTableSummary for a single GSI.

type AdminGroup 

type AdminGroup interface {
	Status() raftengine.Status
	Configuration(ctx context.Context) (raftengine.Configuration, error)
}

AdminGroup exposes per-Raft-group state to the Admin service. It is a narrow subset of raftengine.Engine so tests can supply an in-memory fake without standing up a real Raft cluster. Configuration is polled on each GetClusterOverview to pick up scale-out / scale-in events without the operator having to restart the admin binary.

type AdminItem 

type AdminItem struct {
	Attributes map[string]AdminAttributeValue `json:"attributes"`
}

AdminItem is the admin-facing projection of one DynamoDB item.

type AdminItemKey 

type AdminItemKey struct {
	PartitionKey AdminAttributeValue  `json:"partition_key"`
	SortKey      *AdminAttributeValue `json:"sort_key,omitempty"`
}

AdminItemKey carries the primary key of a single item; SortKey is nil for hash-only tables.

type AdminListObjectsOptions 

type AdminListObjectsOptions struct {
	Prefix            string `json:"prefix,omitempty"`
	Delimiter         string `json:"delimiter,omitempty"`
	ContinuationToken string `json:"continuation_token,omitempty"`
	MaxKeys           int    `json:"max_keys,omitempty"`
}

AdminListObjectsOptions controls one AdminListObjects call. Defaults match the design doc §3.1.2: MaxKeys=100, clamped to [1, adminListObjectsMaxKeys=1000].

type AdminObject 

type AdminObject struct {
	Key          string    `json:"key"`
	Size         int64     `json:"size"`
	ContentType  string    `json:"content_type"`
	ETag         string    `json:"etag"`
	LastModified time.Time `json:"last_modified"`
	StorageClass string    `json:"storage_class"`
}

AdminObject is the admin-facing metadata projection. Mirrors the fields the design doc §3.1.2 names — no internal-only state (uploadID, chunk sizes, etc.) leaks to admin handlers / SPA.

type AdminObjectListing 

type AdminObjectListing struct {
	Objects               []AdminObject `json:"objects"`
	CommonPrefixes        []string      `json:"common_prefixes,omitempty"`
	NextContinuationToken string        `json:"next_continuation_token,omitempty"`
}

AdminObjectListing is the AdminListObjects response shape. CommonPrefixes are populated only when Options.Delimiter is set; they represent the pseudo-directories the SPA renders as foldable rows. NextContinuationToken is empty when the page fully drained the prefix scan.

type AdminPeekMessageOptions 

type AdminPeekMessageOptions struct {
	// Limit caps the number of messages returned. Clamped to
	// [1, adminPeekMaxLimit]; 0 means "use default
	// (adminPeekDefaultLimit)".
	Limit int
	// Cursor is an opaque continuation token from a prior call;
	// empty means "start from the front of the visibility index".
	Cursor string
	// BodyMaxBytes truncates message bodies at this length to
	// bound response size. Clamped to
	// [adminPeekMinBodyBytes, sqsMaximumAllowedMaximumMessageSize]
	// (= 256 KiB, matching AWS SQS's hard cap on stored message
	// size). 0 means "use default (adminPeekDefaultBodyBytes)".
	// The full body is always retained on the server; only the
	// wire representation is truncated.
	BodyMaxBytes int
}

AdminPeekMessageOptions controls a peek call. Zero values map to the documented defaults: Limit=20, Cursor=empty, BodyMaxBytes=4096.

type AdminPeekedAttribute 

type AdminPeekedAttribute struct {
	DataType    string `json:"data_type"`
	StringValue string `json:"string_value,omitempty"`
	// BinaryValue carries the raw bytes; the JSON wire form
	// base64-encodes (standard Go encoding/json behaviour for
	// []byte) so binary payloads survive the SPA round-trip.
	BinaryValue []byte `json:"binary_value,omitempty"`
}

AdminPeekedAttribute mirrors the typed shape SQS uses for MessageAttribute values — DataType (e.g. "String", "Number", "Binary", "String.MyCustom") plus the value in the appropriate representation. The earlier draft used map[string]string here, which would have flattened the typed attribute set stored in sqsMessageRecord.MessageAttributes and silently dropped binary payloads + the DataType discriminator (Codex r11 on the design doc). Operators triaging a DLQ need both — a message routed there because of an attribute-encoding mismatch is invisible if peek only surfaces stringified values.

type AdminPeekedMessage 

type AdminPeekedMessage struct {
	MessageID        string                          `json:"message_id"`
	Body             string                          `json:"body"`                       // truncated per opts.BodyMaxBytes
	BodyTruncated    bool                            `json:"body_truncated"`             // true when Body was cut
	BodyOriginalSize int64                           `json:"body_original_size"`         // bytes in the original body, for display
	SentTimestamp    time.Time                       `json:"sent_timestamp"`             // SQS SentTimestamp
	ReceiveCount     int32                           `json:"receive_count"`              // ApproximateReceiveCount
	GroupID          string                          `json:"group_id,omitempty"`         // FIFO MessageGroupId, empty for standard
	DeduplicationID  string                          `json:"deduplication_id,omitempty"` // FIFO MessageDeduplicationId, empty for standard
	Attributes       map[string]AdminPeekedAttribute `json:"attributes,omitempty"`       // typed SQS message attributes
}

AdminPeekedMessage is one row in the peek result. JSON tags pin the snake_case wire shape the design doc §3.5 specifies; without them the encoder would emit Go-style PascalCase field names and the SPA's client adapter would silently misparse every row. CodeRabbit r4 caught the regression.

type AdminPrincipal 

type AdminPrincipal struct {
	AccessKey string
	Role      AdminRole
}

AdminPrincipal is the authentication context every admin write entrypoint takes. The adapter re-evaluates authorisation against this principal *itself* — it does not trust the caller to have already enforced the role. That is the design's "認可の真実は常に adapter 側" invariant (Section 3.2): if a follower forwards a pre-authenticated request via the future AdminForward RPC, the leader must still verify before acting.

type AdminPurgeResult 

type AdminPurgeResult struct {
	GenerationBefore uint64
	GenerationAfter  uint64
}

AdminPurgeResult is the success return of AdminPurgeQueue. The generation values are captured from the committed OCC round, not a separate pre/post meta read — a second loadQueueMetaAt call would race a concurrent purge resetting LastPurgedAtMillis in the 60-second window and could log generation values that never appeared as a single consistent state.

type AdminQueueCounters 

type AdminQueueCounters struct {
	Visible    int64
	NotVisible int64
	Delayed    int64
}

AdminQueueCounters matches sqsApproxCounters (int64) so the admin bridge does not have to convert between widths. Visible / NotVisible / Delayed are the AWS Approximate* triple.

type AdminQueueSummary 

type AdminQueueSummary struct {
	Name       string
	IsFIFO     bool
	Generation uint64
	CreatedAt  time.Time
	Attributes map[string]string
	Counters   AdminQueueCounters
	// IsDLQ is true when at least one other queue's RedrivePolicy
	// resolves its deadLetterTargetArn to this queue. The SPA uses
	// the flag to switch the Messages-tab framing and the Purge
	// button label between "Purge messages" and "Purge DLQ".
	IsDLQ bool
	// DLQSources lists the names of queues whose RedrivePolicy
	// points at this queue, in lexicographic order. Empty when
	// IsDLQ is false. Used by the SPA to render confirmation copy
	// like "This queue is the DLQ for orders, payments". The
	// computation is a paginated reverse scan over the queue-meta
	// prefix at the same read timestamp as the meta load; values
	// reflect the same MVCC snapshot the rest of the summary does.
	DLQSources []string
}

AdminQueueSummary is the per-queue projection the admin dashboard surfaces. It deliberately covers only the fields the SPA renders so the package's wire-format types stay internal.

Counters mirror the AWS Approximate* attribute set produced by scanApproxCounters; they are best-effort by AWS contract and stop counting once the catalog's per-call cap is reached (the SPA polls continuously, so an unbounded scan would pin the leader).

type AdminRole 

type AdminRole string

AdminRole is the authorization tier the adapter checks against on every admin write entrypoint. The constants intentionally mirror internal/admin.Role string values so the wire / persisted role vocabulary stays aligned across packages, but we keep a separate type here so the adapter has zero dependency on internal/admin. 

const (
	// AdminRoleReadOnly may issue list / describe but not create or delete.
	AdminRoleReadOnly AdminRole = "read_only"
	// AdminRoleFull may issue every admin operation.
	AdminRoleFull AdminRole = "full"
)

type AdminScanOptions 

type AdminScanOptions struct {
	Limit          int                            `json:"limit,omitempty"`
	ExclusiveStart map[string]AdminAttributeValue `json:"exclusive_start,omitempty"`
}

AdminScanOptions controls one AdminScanTable call. Defaults match the design doc §3.1.1: 

Limit = 25 (clamped to [1, adminItemScanMaxLimit=100])
StartKey = nil (front of the table)

type AdminScanResult 

type AdminScanResult struct {
	Items            []AdminItem                    `json:"items"`
	LastEvaluatedKey map[string]AdminAttributeValue `json:"last_evaluated_key,omitempty"`
}

AdminScanResult is the admin-package projection of the underlying scan output: the bounded items page plus a continuation key (nil when the scan has drained for the current MVCC snapshot).

type AdminServer 

type AdminServer struct {
	pb.UnimplementedAdminServer
	// contains filtered or unexported fields
}

AdminServer implements the node-side Admin gRPC service described in docs/admin_ui_key_visualizer_design.md §4 (Layer A). Phase 0 only implements GetClusterOverview and GetRaftGroups; remaining RPCs return Unimplemented so the generated client can still compile against older nodes during rollout.

func NewAdminServer 

func NewAdminServer(self NodeIdentity, members []NodeIdentity) *AdminServer

NewAdminServer constructs an AdminServer. `self` identifies the local node for responses that return node identity. `members` is the static membership snapshot shipped to the admin binary; callers that already have a membership source may pass nil and let the admin binary's fan-out layer discover peers by other means.

func (*AdminServer) GetClusterOverview 

func (s *AdminServer) GetClusterOverview(
	ctx context.Context,
	_ *pb.GetClusterOverviewRequest,
) (*pb.GetClusterOverviewResponse, error)

GetClusterOverview returns the local node identity, the current member list, and per-group leader identity collected from the engines registered via RegisterGroup. The member list is the union of (a) the bootstrap seed supplied to NewAdminServer and (b) the live Configuration of every registered Raft group — the latter picks up scale-out nodes added after startup so the admin binary's fan-out discovery does not miss them.

func (*AdminServer) GetKeyVizMatrix 

func (s *AdminServer) GetKeyVizMatrix(
	_ context.Context,
	req *pb.GetKeyVizMatrixRequest,
) (*pb.GetKeyVizMatrixResponse, error)

GetKeyVizMatrix renders the keyviz heatmap matrix for the [from, to) range supplied by the request, returning one KeyVizRow per tracked route or virtual bucket and a parallel column-timestamp slice.

Series selection (Reads / Writes / ReadBytes / WriteBytes) maps from the request's KeyVizSeries enum to the matching keyviz.MatrixRow counter; KEYVIZ_SERIES_UNSPECIFIED defaults to Reads.

Returns codes.Unavailable when no sampler is registered (keyviz disabled) so callers can distinguish that from "no data yet" (which yields a successful empty response).

func (*AdminServer) GetRaftGroups 

func (s *AdminServer) GetRaftGroups(
	_ context.Context,
	_ *pb.GetRaftGroupsRequest,
) (*pb.GetRaftGroupsResponse, error)

GetRaftGroups returns per-group state snapshots. Phase 0 wires commit/applied indices only; per-follower contact and term history land in later phases.

func (*AdminServer) RegisterGroup 

func (s *AdminServer) RegisterGroup(groupID uint64, g AdminGroup)

RegisterGroup binds a Raft group ID to its engine so the Admin service can report leader and log state for that group.

func (*AdminServer) RegisterSampler 

func (s *AdminServer) RegisterSampler(sampler KeyVizSampler)

RegisterSampler wires the keyviz sampler used by GetKeyVizMatrix. Without this call (or with a nil sampler) the RPC returns codes.Unavailable so callers can distinguish "keyviz disabled" from "no data yet".

func (*AdminServer) SetCapability 

func (s *AdminServer) SetCapability(name string, enabled bool)

SetCapability exposes a local binary or runtime capability in GetClusterOverview. Disabled capabilities are kept in the map so fan-out callers can distinguish "known false" from an older server that lacks the field or key entirely.

func (*AdminServer) SetClock 

func (s *AdminServer) SetClock(now func() time.Time)

SetClock overrides the clock used by GetRaftGroups, letting tests inject a fixed time without mutating any package-global state. Concurrent RPCs on other AdminServer instances are unaffected.

type AdminTableSummary 

type AdminTableSummary struct {
	Name                   string
	PartitionKey           string
	SortKey                string
	Generation             uint64
	GlobalSecondaryIndexes []AdminGSISummary
}

AdminTableSummary is the table-level information the admin dashboard surfaces for a single Dynamo-compatible table. It deliberately projects only the fields the dashboard needs so the package's wire-format types (dynamoTableSchema and friends) stay internal.

type CapabilityFanoutFn 

type CapabilityFanoutFn func(ctx context.Context) (admin.CapabilityFanoutResult, error)

CapabilityFanoutFn is the closure the server invokes to run the §4 Voters ∪ Learners pre-flight before the cutover proposal. Production wiring composes it from internal/admin.CapabilityFanout(routes, dial, timeout) where routes is built from the Raft engine's live membership view and dial reuses the existing admin connection pool. Tests stub it with a deterministic result to exercise the §4.3 OK / refuse branches at the RPC layer without spinning real clients.

type CutoverBarrierController 

type CutoverBarrierController interface {
	// Begin opens the barrier on every participating ShardGroup.
	// Returns a channel that closes when in-flight drains. The
	// typical caller uses WaitDrained instead so context cancellation
	// composes; the channel is exposed so callers MAY use a select
	// with other signals when needed.
	Begin() <-chan struct{}
	// WaitDrained blocks until in-flight drains or ctx fires.
	// Returns nil on drain, wrapped ctx.Err() on cancellation. A
	// barrier-incapable controller (test fixture) may degrade to
	// immediate-success; production controllers MUST honour the
	// drain semantic so the handler doesn't propose the cutover
	// while user proposals are still landing.
	WaitDrained(ctx context.Context) error
	// InstallWrap publishes the active raft envelope wrap closure
	// on every participating ShardGroup, in step 5 of the barrier
	// sequence (after the cutover entry has both committed AND
	// applied locally). The closure source is owned by the
	// controller, not the EncryptionAdminServer — 6E-2e populates
	// it via main.go wiring from the sidecar's Active.Raft DEK.
	InstallWrap()
	// End closes the barrier on every participating ShardGroup.
	// Idempotent against double-End. Pair with Begin via defer in
	// the handler.
	End()
}

CutoverBarrierController coordinates the §7.1 quiescence barrier across every ShardGroup that participates in the raft-envelope cutover. EnableRaftEnvelope drives the 6-step sequence: 

Begin()          // step 1: block USER proposals
WaitDrained(ctx) // step 2: drain in-flight
<handler proposes cutover entry via ProposeAdmin>
<handler waits for FSM apply via latestAppliedIndex>
InstallWrap()    // step 5: SetRaftPayloadWrap on each group
End()            // step 6: unblock USER proposals

Production wiring (6E-2e: main.go) fans the controller over every participating ShardGroup so each leader's dynamicWrappedProposer.Propose engages the gate. Tests stub the interface to drive the state machine without spinning real engines.

All four methods MUST be safe to call from one handler goroutine at a time (per-handler serialization via cutoverSem); the controller does not need internal locking against re-entry from itself. Cross-goroutine reads of the underlying state (Propose gate check, drain signal) ARE concurrent and rely on the controller's own happens-before ordering.

type DeltaCompactor 

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

DeltaCompactor folds accumulated delta keys into their corresponding base metadata keys for all wide-column collection types (List, Hash, Set, ZSet).

It runs as a background goroutine on the Raft leader. Non-leaders skip each tick silently. Compaction is performed as an OCC transaction so concurrent writers never conflict with the compactor.

func NewDeltaCompactor 

func NewDeltaCompactor(st store.MVCCStore, coord kv.Coordinator, opts ...DeltaCompactorOption) *DeltaCompactor

NewDeltaCompactor creates a DeltaCompactor that operates on st using coord.

func (*DeltaCompactor) Run 

func (c *DeltaCompactor) Run(ctx context.Context) error

Run starts the background compaction loop and blocks until ctx is cancelled.

func (*DeltaCompactor) SyncOnce 

func (c *DeltaCompactor) SyncOnce(ctx context.Context) error

SyncOnce runs one compaction pass. The IsLeader() guard avoids the full-prefix delta scan on followers, which would proxy cross-node on ShardStore backends. For sharded deployments where this node is the leader for a non-default shard group, the regular delta-compaction tick is skipped; those keys are still handled by the urgent compaction path (compactUrgentKey) which uses IsLeaderForKey for per-key routing.

The TTL-inline migrator is intentionally outside the default-group leader guard. It is a migration path, not just a hot-key repair path, and each candidate is still filtered by IsLeaderForKey before dispatch. Without this pass, keys whose shard is led by this node but whose default group is led elsewhere would never get their inline TTL anchor before legacy fallback is disabled.

Collection-type handlers run sequentially under one per-tick timeout. A production leader can otherwise receive four full-prefix scans at the same time; under Redis migration load that can starve Raft heartbeats and cause callers to see connection resets. Urgent single-key compaction and TTL migration remain independently key-routed.

func (*DeltaCompactor) TriggerUrgentCompaction 

func (c *DeltaCompactor) TriggerUrgentCompaction(typeName string, userKey []byte)

TriggerUrgentCompaction queues an immediate single-key compaction for a key whose delta count has exceeded MaxDeltaScanLimit. The request is dropped silently when the channel is full (the regular tick will catch it).

type DeltaCompactorOption 

type DeltaCompactorOption func(*DeltaCompactor)

DeltaCompactorOption configures a DeltaCompactor.

func WithDeltaCompactorLogger 

func WithDeltaCompactorLogger(l *slog.Logger) DeltaCompactorOption

WithDeltaCompactorLogger sets the logger.

func WithDeltaCompactorMaxDeltaCount 

func WithDeltaCompactorMaxDeltaCount(n int) DeltaCompactorOption

WithDeltaCompactorMaxDeltaCount sets the soft threshold at which a key's deltas are folded into its base metadata. Default: 64.

func WithDeltaCompactorScanInterval 

func WithDeltaCompactorScanInterval(d time.Duration) DeltaCompactorOption

WithDeltaCompactorScanInterval sets the period between compaction passes. Default: 30s.

func WithDeltaCompactorTimeout 

func WithDeltaCompactorTimeout(d time.Duration) DeltaCompactorOption

WithDeltaCompactorTimeout sets the per-tick timeout. Default: 5s.

func WithDeltaCompactorTimeoutBackoff 

func WithDeltaCompactorTimeoutBackoff(d time.Duration) DeltaCompactorOption

WithDeltaCompactorTimeoutBackoff sets how long background passes are skipped after a tick exhausts its timeout. Default: 1m.

type DistributionFilesystemObserver 

type DistributionFilesystemObserver interface {
	ObserveFilePinnedHotspot(reason string)
}

type DistributionServer 

type DistributionServer struct {
	pb.UnimplementedDistributionServer
	// contains filtered or unexported fields
}

DistributionServer serves distribution related gRPC APIs.

func NewDistributionServer 

func NewDistributionServer(e *distribution.Engine, catalog *distribution.CatalogStore, opts ...DistributionServerOption) *DistributionServer

NewDistributionServer creates a new server.

func (*DistributionServer) GetCatalogCapabilities 

func (s *DistributionServer) GetCatalogCapabilities(ctx context.Context, _ *pb.CatalogCapabilitiesRequest) (*pb.CatalogCapabilitiesResponse, error)

GetCatalogCapabilities negotiates the durable delta-watch protocol.

func (*DistributionServer) GetRoute 

func (s *DistributionServer) GetRoute(ctx context.Context, req *pb.GetRouteRequest) (*pb.GetRouteResponse, error)

GetRoute returns route for a key.

func (*DistributionServer) GetTimestamp 

func (s *DistributionServer) GetTimestamp(ctx context.Context, req *pb.GetTimestampRequest) (*pb.GetTimestampResponse, error)

GetTimestamp returns monotonically increasing timestamp.

func (*DistributionServer) ListRoutes 

func (s *DistributionServer) ListRoutes(ctx context.Context, req *pb.ListRoutesRequest) (*pb.ListRoutesResponse, error)

ListRoutes returns all durable routes from catalog storage.

func (*DistributionServer) SplitRange 

func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeRequest) (*pb.SplitRangeResponse, error)

SplitRange splits a route into two child routes in the same raft group.

func (*DistributionServer) UpdateRoute 

func (s *DistributionServer) UpdateRoute(start, end []byte, group uint64)

UpdateRoute allows updating route information.

func (*DistributionServer) WatchCatalog 

func (s *DistributionServer) WatchCatalog(req *pb.CatalogWatchRequest, stream pb.Distribution_WatchCatalogServer) error

WatchCatalog streams contiguous deltas and emits a snapshot reset when the requested reconnect cursor predates retained history.

type DistributionServerOption 

type DistributionServerOption func(*DistributionServer)

DistributionServerOption configures DistributionServer behavior.

func WithCatalogReloadRetryPolicy 

func WithCatalogReloadRetryPolicy(attempts int, interval time.Duration) DistributionServerOption

WithCatalogReloadRetryPolicy configures the retry policy used after split commit when waiting for the local catalog snapshot to become visible.

func WithCatalogWatchInterval 

func WithCatalogWatchInterval(interval time.Duration) DistributionServerOption

WithCatalogWatchInterval sets how often an idle catalog stream checks for a newly committed version.

func WithCatalogWatchLeaderCheck 

func WithCatalogWatchLeaderCheck(check func() bool) DistributionServerOption

WithCatalogWatchLeaderCheck restricts long-lived catalog streams to the current default-group leader. Existing callers without a check retain the package-level server behavior used by tests and embedded deployments.

func WithDistributionActiveTimestampTracker 

func WithDistributionActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) DistributionServerOption

func WithDistributionCoordinator 

func WithDistributionCoordinator(coordinator kv.Coordinator) DistributionServerOption

WithDistributionCoordinator configures the coordinator used for Raft-backed catalog mutations in SplitRange.

func WithDistributionFilesystemObserver 

func WithDistributionFilesystemObserver(observer DistributionFilesystemObserver) DistributionServerOption

type DynamoDBServer 

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

func NewDynamoDBServer 

func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordinator, opts ...DynamoDBServerOption) *DynamoDBServer

func (*DynamoDBServer) AdminCreateTable 

func (d *DynamoDBServer) AdminCreateTable(ctx context.Context, principal AdminPrincipal, in AdminCreateTableInput) (*AdminTableSummary, error)

AdminCreateTable creates a Dynamo-compatible table on the local node, after re-validating the principal's role and confirming this node is the verified Raft leader. The returned summary mirrors the shape of AdminDescribeTable on the same name so the SPA can show the freshly-created table without an extra describe round-trip.

Errors:

  • ErrAdminForbidden when the principal cannot write.
  • ErrAdminNotLeader when the node is a follower.
  • The adapter's standard dynamoAPIError chain for validation / storage failures, preserved unmodified so the HTTP handler can map the inner code (ValidationException, ResourceInUseException, etc.) to the appropriate status without re-classifying.

func (*DynamoDBServer) AdminDeleteItem 

func (d *DynamoDBServer) AdminDeleteItem(ctx context.Context, principal AdminPrincipal, tableName string, key map[string]AdminAttributeValue) error

AdminDeleteItem removes one item by primary key. Write role required.

Sentinels:

  • ErrAdminForbidden — principal lacks write role
  • ErrAdminNotLeader — follower
  • ErrAdminDynamoNotFound — table absent
  • ErrAdminDynamoValidation — empty / malformed input

func (*DynamoDBServer) AdminDeleteTable 

func (d *DynamoDBServer) AdminDeleteTable(ctx context.Context, principal AdminPrincipal, name string) error

AdminDeleteTable is the SigV4-bypass counterpart to deleteTable. Returns the same sentinel errors as AdminCreateTable plus the adapter's standard dynamoErrResourceNotFound when the table is absent — admin handlers should map that to 404 rather than 500.

func (*DynamoDBServer) AdminDescribeTable 

func (d *DynamoDBServer) AdminDescribeTable(ctx context.Context, name string) (*AdminTableSummary, bool, error)

AdminDescribeTable returns a schema snapshot for name. The triple (result, present, error) lets admin callers distinguish a genuine "not found" from a storage error without sniffing sentinels: when the table is missing the function returns (nil, false, nil).

Unlike the SigV4 describeTable handler, AdminDescribeTable does NOT invoke ensureLegacyTableMigration. The admin dashboard is a strictly read-only surface (Gemini medium review on PR #633), so triggering Raft-coordinated key-encoding migrations as a side effect of routine polling would (a) violate the read-only contract and (b) cause every dashboard refresh to write to the cluster. Migration still runs lazily on the next SigV4 read or write of the same table — the schema we return here is just a snapshot for display, not a guarantee that the table is up-to-date for serving.

func (*DynamoDBServer) AdminGetItem 

func (d *DynamoDBServer) AdminGetItem(ctx context.Context, principal AdminPrincipal, tableName string, key map[string]AdminAttributeValue) (*AdminItem, bool, error)

AdminGetItem fetches one item by primary key. Returns (nil, false, nil) when the item is absent (not an error).

Sentinels:

  • ErrAdminForbidden — principal lacks read role
  • ErrAdminNotLeader — follower
  • ErrAdminDynamoNotFound — table absent
  • ErrAdminDynamoValidation — empty / malformed input

func (*DynamoDBServer) AdminListTables 

func (d *DynamoDBServer) AdminListTables(ctx context.Context) ([]string, error)

AdminListTables returns every Dynamo-style table this server knows about, in the lexicographic order the metadata index produces. Intended for the in-process admin listener as the SigV4-free counterpart to the listTables HTTP handler; both share the same underlying lookup so the two views cannot drift.

func (*DynamoDBServer) AdminPutItem 

func (d *DynamoDBServer) AdminPutItem(ctx context.Context, principal AdminPrincipal, tableName string, item AdminItem) error

AdminPutItem creates-or-replaces one item. Write role required.

Sentinels:

  • ErrAdminForbidden — principal lacks write role
  • ErrAdminNotLeader — follower
  • ErrAdminDynamoNotFound — table absent
  • ErrAdminDynamoValidation — empty / malformed input

func (*DynamoDBServer) AdminScanTable 

func (d *DynamoDBServer) AdminScanTable(ctx context.Context, principal AdminPrincipal, tableName string, opts AdminScanOptions) (AdminScanResult, error)

AdminScanTable returns a bounded page of items. Read-only.

Sentinels:

  • ErrAdminForbidden — principal lacks read role
  • ErrAdminNotLeader — follower
  • ErrAdminDynamoNotFound — table absent
  • ErrAdminDynamoValidation — empty / malformed input

Scan-response budget: a single underlying scan may return a LastEvaluatedKey before reaching Limit because of DynamoDB's 1 MiB response cap. The admin RPC does NOT loop internally to refill the page; partial pages are documented behaviour and the caller treats any non-nil LastEvaluatedKey as "more available".

func (*DynamoDBServer) Run 

func (d *DynamoDBServer) Run() error

func (*DynamoDBServer) SetListener 

func (d *DynamoDBServer) SetListener(listen net.Listener)

SetListener installs the listener Run serves from. Startup wiring uses this to build admin-forward sources before binding the public socket.

func (*DynamoDBServer) Stop 

func (d *DynamoDBServer) Stop()

type DynamoDBServerOption 

type DynamoDBServerOption func(*DynamoDBServer)

func WithDynamoDBActiveTimestampTracker 

func WithDynamoDBActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) DynamoDBServerOption

func WithDynamoDBLeaderMap 

func WithDynamoDBLeaderMap(m map[string]string) DynamoDBServerOption

WithDynamoDBLeaderMap configures the Raft-address-to-DynamoDB-address mapping used to forward requests from followers to the current leader. The format mirrors the raftRedisMap / raftS3Map convention.

func WithDynamoDBRequestObserver 

func WithDynamoDBRequestObserver(observer monitoring.DynamoDBRequestObserver) DynamoDBServerOption

WithDynamoDBRequestObserver enables Prometheus-compatible request metrics.

func WithDynamoOnePhaseTxnDedup 

func WithDynamoOnePhaseTxnDedup(enabled bool) DynamoDBServerOption

WithDynamoOnePhaseTxnDedup enables or disables the option-2 one-phase idempotency dedup on the single-item write retry path (see DynamoDBServer.onePhaseTxnDedup). It defaults on; pass false to opt out from code, or set ELASTICKV_DYNAMODB_ONEPHASE_DEDUP=0 to opt out from the environment. The constructor option trumps the env var.

type EncryptionAdminServer 

type EncryptionAdminServer struct {
	pb.UnimplementedEncryptionAdminServer
	// contains filtered or unexported fields
}

EncryptionAdminServer implements proto.EncryptionAdmin. Wires the §6.1 read-only probes (GetCapability, GetSidecarState, ResyncSidecar) plus the three mutating opcodes (Bootstrap, RotateDEK, RegisterEncryptionWriter) onto Stage 3's raft envelope and Stage 4's fsmwire body encoders. Mutators are leader-gated through requireLeader / VerifyLeader.

func NewEncryptionAdminServer 

func NewEncryptionAdminServer(opts ...EncryptionAdminServerOption) *EncryptionAdminServer

NewEncryptionAdminServer constructs an EncryptionAdminServer. The returned server is safe to register on the same gRPC listener as Admin / Distribution; every RPC is concurrency-safe because the only mutable state is the sidecar file, which encryption.WriteSidecar already serialises via the §5.1 crash-durable write protocol.

Production wiring MUST call Validate after construction so a configuration that wires a proposer but forgets the leaderView fails closed at startup rather than silently letting followers mutate state.

func (*EncryptionAdminServer) BootstrapEncryption 

func (s *EncryptionAdminServer) BootstrapEncryption(ctx context.Context, req *pb.BootstrapEncryptionRequest) (*pb.BootstrapEncryptionResponse, error)

BootstrapEncryption proposes the §5.6 0x04 OpBootstrap entry that installs the initial wrapped DEK pair AND the cluster-wide writer-registry batch in a single Raft transaction. The server accepts a pre-built writer batch (capability fan-out is the CLI's job today) and validates every entry against the §6.1 invariants before delegating to fsmwire.EncodeBootstrap. FSM apply enforces the idempotency check (rejects if the sidecar's active.storage is already set), so re-running this RPC against an already-bootstrapped cluster is safe — the engine returns the apply error and ErrEncryptionApply halts the apply loop.

func (*EncryptionAdminServer) EnableRaftEnvelope 

func (s *EncryptionAdminServer) EnableRaftEnvelope(ctx context.Context, req *pb.EnableRaftEnvelopeRequest) (*pb.EnableRaftEnvelopeResponse, error)

EnableRaftEnvelope is the Stage 6E Phase 2 cutover — flips Raft proposals from cleartext to §4.2-envelope. Structural mirror of EnableStorageEnvelope; the differences are:

  • Target Purpose is PurposeRaft.
  • Source DEK slot is sidecar.Active.Raft (not Active.Storage).
  • The "already active" sentinel is the single field sidecar.RaftEnvelopeCutoverIndex != 0 — there is no separate bool flag, so the raft variant has no equivalent of the §6.4 cutover_index_unknown defensive fallback (a zero index is exactly the not-active state, not a corrupted-active state, and the 6E-1a applier fail-closes on raftIdx == 0 before ApplyRegistration).

The semaphore, pre-check / fan-out / propose / post-check sequence, and error mapping match the storage variant verbatim; see EnableStorageEnvelope for the full design rationale.

**Gate posture**: raftEnvelopeWrapEnabled remains a test/emergency kill switch. If closed, the pre-gate validation surface (leader, semaphore acquire, request shape) still fires, but no Raft proposal is composed and no sidecar mutation occurs.

func (*EncryptionAdminServer) EnableStorageEnvelope 

func (s *EncryptionAdminServer) EnableStorageEnvelope(ctx context.Context, req *pb.EnableStorageEnvelopeRequest) (*pb.EnableStorageEnvelopeResponse, error)

EnableStorageEnvelope is the §7.1 Phase 1 cutover RPC: it proposes a RotateSubEnableStorageEnvelope (0x04) rotation entry that flips sidecar.StorageEnvelopeActive on every replica simultaneously and stores the original cutover index in sidecar.StorageEnvelopeCutoverIndex (§6.4). After the entry applies, the storage layer's WithStorageEnvelopeGate (Stage 6D-5) reads true on every Put and the cluster begins emitting §4.1 envelopes for new versions.

The RPC composes the §4 Voters ∪ Learners capability fan-out helper (Stage 6D-3), the Stage 6D-4 wire dispatch, and the idempotency contract (§6.4): a duplicate call against an already-active sidecar returns OK with `was_already_active=true` and `applied_index = sidecar.StorageEnvelopeCutoverIndex`. Returning AlreadyExists instead would drop the response body per unary-gRPC semantics, so the idempotency discriminator lives on the success path.

The server-side sequence (per design doc §3.2):

  1. Validate `proposer_node_id != 0` and `proposer_local_epoch <= 0xFFFF` at the gRPC boundary.
  2. Verify Stage 6B mutators are enabled — implicit via `s.proposer == nil` (matches RotateDEK / BootstrapEncryption posture).
  3. Verify we are the default-group leader via `requireLeader`.
  4. Verify the sidecar has Active.Storage != 0 (bootstrap committed) — return FailedPrecondition with a "run BootstrapEncryption first" hint otherwise.
  5. If sidecar.StorageEnvelopeActive == true, return the §3.2 step 5 idempotent-retry response (OK + was_already_active + applied_index = StorageEnvelopeCutoverIndex). Skip the fan-out — the original cutover already passed the gate.
  6. Refuse with FailedPrecondition if the capability fan-out is not wired (`s.capabilityFanout == nil`): the cutover MUST have a §4 pre-flight; a silent skip would let an unreachable learner sneak through.
  7. Run the fan-out. Any verdict with Reachable=false or EncryptionCapable=false refuses with FailedPrecondition; the response detail names the specific node.
  8. Compose the RotationPayload (§2.1: empty Wrapped, DEKID = sidecar.Active.Storage, Purpose = PurposeStorage, ProposerRegistration covering the active storage DEK).
  9. Propose through Raft via `proposeEncryptionEntry`.
  10. Re-read the sidecar to discriminate fresh-success vs. stale-DEKID race vs. concurrent-overlap idempotent and assemble the response.

FSM-level no-op outcomes (stale DEKID via a RotateDEK race, already-active via a duplicate cutover) do NOT halt the apply path — the 6D-4 applier deliberately consumes those entries without flipping the sidecar field. The RPC discriminates by reading the post-apply sidecar: still false ⇒ stale DEKID, surface as FailedPrecondition with the §2.1 #3 retry hint; now true with cutover-index mismatch ⇒ another cutover landed concurrently, treat as idempotent success.

func (*EncryptionAdminServer) GetCapability 

func (s *EncryptionAdminServer) GetCapability(_ context.Context, _ *pb.Empty) (*pb.CapabilityReport, error)

GetCapability returns the §6.1 CapabilityReport for the local node. The RPC is read-only and side-effect-free; it must be servable on every node regardless of leadership state so the §7.1 cutover command can fan it out across voters and learners.

encryption_capable is gated on "this node was restarted with --encryption-enabled" (proxied here by a non-empty sidecar path), NOT on whether the sidecar has been bootstrapped. The §7.1 Phase 0 capability gate fires before bootstrap proposes the first DEK, so gating on Active.Storage != 0 would deadlock the cutover against the very entry it is gating. SidecarPresent then carries the orthogonal "has the bootstrap entry landed yet" signal.

A node that has never been started with --encryption-enabled (empty sidecarPath) returns encryption_capable=false with local_epoch=0 and full_node_id=0 per the §6.1 contract: the cutover command refuses with ErrCapabilityCheckFailed in that case so the empty epoch never reaches the writer registry.

func (*EncryptionAdminServer) GetSidecarState 

func (s *EncryptionAdminServer) GetSidecarState(ctx context.Context, _ *pb.Empty) (*pb.SidecarStateReport, error)

GetSidecarState returns the §5.5 compaction-fallback snapshot. The response includes every unretired wrapped DEK plus the two active pointers; the wrapped material is leakage-safe because it is KEK-wrapped, which is the same property the on-disk sidecar has.

GetSidecarState has no caller-id request field, so the writer_registry_for_caller map is populated for this server's own fullNodeID only when both fullNodeID and writerRegistry are wired and the registry leader view can satisfy an applied read. Followers still return their local sidecar snapshot with an empty projection so encryption status remains available from any node. The follower-repair flow that needs a remote caller projection uses ResyncSidecar, whose request carries caller_full_node_id.

func (*EncryptionAdminServer) RegisterEncryptionWriter 

func (s *EncryptionAdminServer) RegisterEncryptionWriter(ctx context.Context, req *pb.RegisterEncryptionWriterRequest) (*pb.RegisterEncryptionWriterResponse, error)

RegisterEncryptionWriter proposes a §11.3 0x03 OpRegistration entry for the calling node's first encrypted-write epoch under the supplied dek_id. The proto carries `repeated WriterBatch` for forward-compatibility, but each PR-B call MUST provide exactly one entry; batched registrations belong to the §5.6 step 1a bootstrap path (deferred to PR-C). The server enforces the single-entry contract here so an operator who accidentally fans out a batched RegisterEncryptionWriter does not silently propose a fraction of it.

func (*EncryptionAdminServer) ResyncSidecar 

func (s *EncryptionAdminServer) ResyncSidecar(ctx context.Context, req *pb.ResyncSidecarRequest) (*pb.ResyncSidecarResponse, error)

ResyncSidecar is the §5.5 follower-repair RPC. The follower asks the leader for the current wrapped DEK set so it can rewrite a sidecar that fell behind a Raft-log compaction window. The RPC is read-only on the server side; no Raft proposal is involved.

Leader-only via requireRegistryLeader, which calls LinearizableRead(ctx) to confirm leadership and wait until the returned ReadIndex has applied locally before registry projections are read. The handler also repeats the applied fence after loading keys.json, because bootstrap and rotation apply writes the sidecar before the matching writer-registry rows.

func (*EncryptionAdminServer) RotateDEK 

func (s *EncryptionAdminServer) RotateDEK(ctx context.Context, req *pb.RotateDEKRequest) (*pb.RotateDEKResponse, error)

RotateDEK proposes a §5.2 rotation as a §11.3 0x05 OpRotation entry. The server validates purpose / dek_id / local_epoch boundaries at the gRPC boundary (the §6.1 doc rule for local_epoch <= 0xFFFF on the wire) and the FSM apply layer enforces the sidecar-level invariants (no rotation while a previous rotation is in flight, etc). Returns the commit index once the entry is durable on a Raft quorum.

func (*EncryptionAdminServer) Validate 

func (s *EncryptionAdminServer) Validate() error

Validate enforces the option-pairing invariants the constructor cannot express through Go's option signature. Tests that wire a proposer without a leaderView are intentionally allowed (the requireLeader path treats nil-leaderView as always-leader for the proposer-wiring unit tests); production wiring in main.go MUST call this method so a misconfiguration fails closed.

type EncryptionAdminServerOption 

type EncryptionAdminServerOption func(*EncryptionAdminServer)

EncryptionAdminServerOption configures EncryptionAdminServer behavior.

func WithEncryptionAdminBuildSHA 

func WithEncryptionAdminBuildSHA(sha string) EncryptionAdminServerOption

WithEncryptionAdminBuildSHA overrides the auto-detected runtime/debug build SHA. Tests use this to pin a deterministic value; production wiring leaves it empty.

func WithEncryptionAdminCapabilityFanout 

func WithEncryptionAdminCapabilityFanout(fn CapabilityFanoutFn) EncryptionAdminServerOption

WithEncryptionAdminCapabilityFanout wires the §4 Voters ∪ Learners pre-flight that the §7.1 Phase 1 cutover RPC (EnableStorageEnvelope) runs before composing the RotateSubEnableStorageEnvelope payload. Without this option EnableStorageEnvelope refuses with FailedPrecondition — matching the proposer / leaderView posture for the other mutator RPCs. A nil argument is a no-op (the server stays in the cutover-disabled posture), mirroring the WithEncryptionAdmin* convention.

func WithEncryptionAdminCutoverBarrier 

func WithEncryptionAdminCutoverBarrier(c CutoverBarrierController) EncryptionAdminServerOption

WithEncryptionAdminCutoverBarrier wires the §7.1 quiescence barrier controller used by EnableRaftEnvelope. A nil argument is a no-op (the server stays in the cutover-disabled posture); EnableRaftEnvelope refuses with FailedPrecondition until both raftEnvelopeWrapEnabled is open AND this controller is wired.

Production wiring (6E-2e: main.go) fans the controller over every ShardGroup that participates in the cutover. Tests pass a stub implementation to exercise the state machine deterministically.

func WithEncryptionAdminFullNodeID 

func WithEncryptionAdminFullNodeID(id uint64) EncryptionAdminServerOption

WithEncryptionAdminFullNodeID sets the 64-bit node id reported in CapabilityReport.full_node_id. The §5.6 step 1a batch keys writer registry entries on (dek_id, uint16(full_node_id)); the leader derives the uint16 narrowing itself, so the server-side value stays at the full 64-bit precision.

func WithEncryptionAdminLatestAppliedIndex 

func WithEncryptionAdminLatestAppliedIndex(fn func() uint64) EncryptionAdminServerOption

WithEncryptionAdminLatestAppliedIndex registers a callback the server uses to populate SidecarStateReport.latest_applied_index and ResyncSidecarResponse.leader_latest_applied_index. A nil callback (the default) reports the value persisted in the sidecar, which lags the in-memory apply counter; the callback is the §5.5 escape hatch for callers that want the freshest value.

func WithEncryptionAdminLeaderView 

func WithEncryptionAdminLeaderView(v raftengine.LeaderView) EncryptionAdminServerOption

WithEncryptionAdminLeaderView registers the leadership oracle. Mutating RPCs and ResyncSidecar reject on followers with FailedPrecondition; the leader's id and address are embedded in the status detail so the operator's CLI can retry against the right node without parsing free-form error text.

func WithEncryptionAdminPostCutoverProposer 

func WithEncryptionAdminPostCutoverProposer(p raftengine.Proposer) EncryptionAdminServerOption

WithEncryptionAdminPostCutoverProposer registers the wrap-aware proposer used for non-cutover-marker admin entries. A nil argument is a no-op, and NewEncryptionAdminServer falls back to the raw proposer when this option is not supplied, preserving the Stage 6B mutator wiring contract.

func WithEncryptionAdminProposer 

func WithEncryptionAdminProposer(p raftengine.Proposer) EncryptionAdminServerOption

WithEncryptionAdminProposer registers the raftengine.Proposer the server uses to propose the §11.3 0x03 / 0x04 / 0x05 entries (Stage 4 wire format). An unset proposer makes every mutating RPC return FailedPrecondition with "proposer not configured", preserving the PR-A production-inert guarantee until Stage 6 flips the cluster flag.

func WithEncryptionAdminRegistryLeaderView 

func WithEncryptionAdminRegistryLeaderView(v raftengine.LeaderView) EncryptionAdminServerOption

WithEncryptionAdminRegistryLeaderView registers the leadership oracle used before sidecar-state RPCs read writer-registry projections. Multi-group production wiring uses the default group leader view here so a non-default group listener cannot serve projection data from a stale default-group follower.

func WithEncryptionAdminSidecarPath 

func WithEncryptionAdminSidecarPath(path string) EncryptionAdminServerOption

WithEncryptionAdminSidecarPath sets the §5.1 keys.json path the server reads on every GetCapability / GetSidecarState / ResyncSidecar call. An empty path means "encryption is not configured on this node"; GetCapability then returns encryption_capable=false instead of erroring.

func WithEncryptionAdminWriterRegistry 

func WithEncryptionAdminWriterRegistry(reg encryption.WriterRegistryStore) EncryptionAdminServerOption

WithEncryptionAdminWriterRegistry wires the durable writer-registry store used by §5.5 sidecar-resync responses to report the leader-recorded local_epoch for the requesting node. A nil value preserves the older read-only posture and returns an empty map.

type GRPCServer 

type GRPCServer struct {
	pb.UnimplementedRawKVServer
	pb.UnimplementedTransactionalKVServer
	// contains filtered or unexported fields
}

func NewGRPCServer 

func NewGRPCServer(store store.MVCCStore, coordinate kv.Coordinator, opts ...GRPCServerOption) *GRPCServer

func (*GRPCServer) Close 

func (r *GRPCServer) Close() error

func (*GRPCServer) Commit 

func (r *GRPCServer) Commit(ctx context.Context, req *pb.CommitRequest) (*pb.CommitResponse, error)

func (*GRPCServer) Delete 

func (r *GRPCServer) Delete(ctx context.Context, req *pb.DeleteRequest) (*pb.DeleteResponse, error)

func (*GRPCServer) Get 

func (r *GRPCServer) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetResponse, error)

func (*GRPCServer) PreWrite 

func (r *GRPCServer) PreWrite(ctx context.Context, req *pb.PreWriteRequest) (*pb.PreCommitResponse, error)

func (*GRPCServer) Put 

func (r *GRPCServer) Put(ctx context.Context, req *pb.PutRequest) (*pb.PutResponse, error)

func (*GRPCServer) RawDelete 

func (r *GRPCServer) RawDelete(ctx context.Context, req *pb.RawDeleteRequest) (*pb.RawDeleteResponse, error)

func (*GRPCServer) RawGet 

func (r *GRPCServer) RawGet(ctx context.Context, req *pb.RawGetRequest) (*pb.RawGetResponse, error)

func (*GRPCServer) RawLatestCommitTS 

func (r *GRPCServer) RawLatestCommitTS(ctx context.Context, req *pb.RawLatestCommitTSRequest) (*pb.RawLatestCommitTSResponse, error)

func (*GRPCServer) RawPut 

func (r *GRPCServer) RawPut(ctx context.Context, req *pb.RawPutRequest) (*pb.RawPutResponse, error)

func (*GRPCServer) RawScanAt 

func (r *GRPCServer) RawScanAt(ctx context.Context, req *pb.RawScanAtRequest) (*pb.RawScanAtResponse, error)

func (*GRPCServer) Rollback 

func (r *GRPCServer) Rollback(ctx context.Context, req *pb.RollbackRequest) (*pb.RollbackResponse, error)

func (*GRPCServer) Scan 

func (r *GRPCServer) Scan(ctx context.Context, req *pb.ScanRequest) (*pb.ScanResponse, error)

type GRPCServerOption 

type GRPCServerOption func(*GRPCServer)

func WithCloseStore 

func WithCloseStore() GRPCServerOption

type HTFIFOCapabilityPeerStatus 

type HTFIFOCapabilityPeerStatus struct {
	// Address is the peer's host:port as supplied to the poller.
	Address string

	// HasHTFIFO is true iff the peer's /sqs_health JSON body's
	// capabilities array contained the htfifo capability string.
	HasHTFIFO bool

	// Capabilities is the parsed capabilities array. Nil on any
	// failure before JSON parsing, or non-nil but missing
	// htfifo when the peer is on an older binary.
	Capabilities []string

	// Error is empty on a clean success (HTTP 200 + parseable
	// JSON, regardless of whether HasHTFIFO is true) and non-empty
	// on any failure (transport error, non-200 status, malformed
	// JSON, or context cancellation).
	Error string
}

HTFIFOCapabilityPeerStatus is one peer's polling result.

type HTFIFOCapabilityReport 

type HTFIFOCapabilityReport struct {
	// AllAdvertise is true iff every peer in the input list
	// returned a /sqs_health body whose `capabilities` array
	// contains the htfifo capability string. False on any timeout,
	// HTTP error, malformed body, or missing-capability — the
	// gate fails closed.
	//
	// Vacuously true on an empty peer list. The caller (CreateQueue
	// gate) is responsible for ensuring the peer list reflects the
	// current cluster membership before consulting this report.
	AllAdvertise bool

	// Peers is the per-peer status, indexed in input order. Each
	// entry has either HasHTFIFO=true (peer advertised the
	// capability) or a non-empty Error explaining why the peer
	// did not pass. Capabilities is the raw list returned by the
	// peer when the body was parseable.
	Peers []HTFIFOCapabilityPeerStatus
}

HTFIFOCapabilityReport summarises the result of polling each peer's /sqs_health endpoint for the htfifo capability. Used by the CreateQueue capability gate (Phase 3.D PR 5) and by operator tooling that needs to confirm a rolling upgrade has finished before enabling partitioned FIFO queues.

AllAdvertise is the binary go/no-go signal for the gate; Peers carries per-node detail for log lines and operator triage.

func PollSQSHTFIFOCapability 

func PollSQSHTFIFOCapability(ctx context.Context, peers []string, cfg PollerConfig) *HTFIFOCapabilityReport

PollSQSHTFIFOCapability polls each peer's /sqs_health endpoint concurrently and reports whether all advertise htfifo. The helper is stateless — every call dials its peers fresh, so a transient network blip on one call does not poison subsequent calls.

Per-peer behaviour:

  • GET http://<peer>/sqs_health with Accept: application/json
  • Expect HTTP 200 and a parseable JSON body matching {"status":"ok","capabilities":[...]}.
  • HasHTFIFO is the membership of htfifo in capabilities.
  • Any failure (transport error, non-200, malformed JSON, timeout, context cancellation) records the reason in Error and leaves HasHTFIFO=false. The poller never returns a fatal error from PollSQSHTFIFOCapability itself; the report carries every per-peer outcome instead.

Concurrency: peers are polled in goroutines; results land via an indexed channel so the slice writes are obviously race-free.

Timeouts: each peer poll is bounded by min(ctx.Deadline(), now+cfg.PerPeerTimeout). A long ctx deadline does not extend the per-peer cap, and an absent ctx deadline still triggers fail-closed at the per-peer cap.

type Internal 

type Internal struct {
	pb.UnimplementedInternalServer
	// contains filtered or unexported fields
}

func NewInternalWithEngine 

func NewInternalWithEngine(txm kv.Transactional, leader raftengine.LeaderView, clock *kv.HLC, relay *RedisPubSubRelay, opts ...InternalOption) *Internal

func (*Internal) Forward 

func (i *Internal) Forward(ctx context.Context, req *pb.ForwardRequest) (*pb.ForwardResponse, error)

func (*Internal) RelayPublish 

func (i *Internal) RelayPublish(_ context.Context, req *pb.RelayPublishRequest) (*pb.RelayPublishResponse, error)

type InternalOption 

type InternalOption func(*Internal)

func WithInternalTimestampAllocator 

func WithInternalTimestampAllocator(alloc kv.TimestampAllocator) InternalOption

type KeyVizSampler 

type KeyVizSampler interface {
	// Snapshot returns the matrix columns in [from, to). Either
	// bound may be the zero time meaning unbounded on that side.
	// Implementations must return rows the caller can mutate freely
	// (a deep copy) — see keyviz.MemSampler.Snapshot.
	Snapshot(from, to time.Time) []keyviz.MatrixColumn
}

KeyVizSampler is the read-side abstraction the Admin service needs from the keyviz package: a time-bounded matrix snapshot. Defined here (not in keyviz) so tests can pass an in-memory fake without constructing a full *keyviz.MemSampler. *keyviz.MemSampler satisfies this interface.

type Node 

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

type NodeIdentity 

type NodeIdentity struct {
	NodeID      string
	GRPCAddress string
}

NodeIdentity is the value form of the protobuf NodeIdentity message used for AdminServer configuration. It avoids copying pb.NodeIdentity, which embeds a protoimpl.MessageState (and a mutex).

type PollerConfig 

type PollerConfig struct {
	// HTTPClient is the client used for /sqs_health GETs. Nil
	// falls back to http.DefaultClient. Callers wanting connection
	// pooling, custom Transport, or shorter Client.Timeout pass
	// their own.
	HTTPClient *http.Client

	// PerPeerTimeout caps how long any single peer's poll runs
	// before being abandoned. Zero defaults to
	// defaultSQSCapabilityPollTimeout (3s). Tests pass a small
	// value (e.g. 100ms) so the per-peer cap path can be
	// exercised quickly without a parent context deadline.
	PerPeerTimeout time.Duration
}

PollerConfig tunes PollSQSHTFIFOCapability for a specific call site. All fields are optional — the zero value picks safe defaults. Tests use the explicit PerPeerTimeout to exercise the per-peer cap independently of any caller-supplied context deadline.

type PurgeInProgressError 

type PurgeInProgressError struct {
	RetryAfter time.Duration
}

PurgeInProgressError is the typed admin error returned by AdminPurgeQueue when the meta-stored 60-second rate limit is active. RetryAfter carries the wall-clock duration the caller should wait, derived from the same LastPurgedAtMillis value the rate-limit check itself read inside the OCC transaction.

func (*PurgeInProgressError) Error 

func (e *PurgeInProgressError) Error() string

func (*PurgeInProgressError) Is 

func (e *PurgeInProgressError) Is(target error) bool

Is implements errors.Is so handlers can sniff ErrAdminSQSPurgeInProgress while still extracting RetryAfter via errors.As. Standard errors-wrapper pattern shared with other typed admin errors in the package.

type RedisApplyObserver 

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

RedisApplyObserver wakes blocking Redis waiters from the FSM apply path. It is safe to share one instance between the FSM wiring and RedisServer.

func NewRedisApplyObserver 

func NewRedisApplyObserver() *RedisApplyObserver

func (*RedisApplyObserver) OnApply 

func (o *RedisApplyObserver) OnApply(op pb.Op, key []byte)

type RedisPubSubRelay 

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

RedisPubSubRelay lets the internal gRPC service publish into the local Redis pubsub bus without depending on RedisServer startup order.

func NewRedisPubSubRelay 

func NewRedisPubSubRelay() *RedisPubSubRelay

func (*RedisPubSubRelay) Bind 

func (r *RedisPubSubRelay) Bind(publish func(channel, message []byte) int64)

func (*RedisPubSubRelay) Publish 

func (r *RedisPubSubRelay) Publish(channel, message []byte) int64

type RedisServer 

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

func NewRedisServer 

func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore, coordinate kv.Coordinator, leaderRedis map[string]string, relay *RedisPubSubRelay, opts ...RedisServerOption) *RedisServer

func (*RedisServer) Close 

func (r *RedisServer) Close() error

Close cancels the base context, signalling all in-flight handlers to abort. Idempotent. The underlying redcon listener is still owned by the caller; Close does NOT touch it so shutdown orchestration can remain with the server owner.

func (*RedisServer) RegisterLuaPoolMetrics 

func (r *RedisServer) RegisterLuaPoolMetrics(registerer prometheus.Registerer) error

RegisterLuaPoolMetrics wires this server's bounded Lua VM pool into the supplied Prometheus registerer, exposing five metrics (hits / misses / drops / idle / max_idle). See monitoring.RegisterLuaPool for the per-metric definitions.

Returns nil if r, the pool, or registerer is nil — callers can invoke this unconditionally from main.go without guarding for test fixtures. The registration uses prometheus.NewCounterFunc / NewGaugeFunc, so the values are read from the pool's atomic counters at scrape time; no observability load is added to the EVAL hot path.

func (*RedisServer) Run 

func (r *RedisServer) Run() error

func (*RedisServer) Stop 

func (r *RedisServer) Stop()

type RedisServerOption 

type RedisServerOption func(*RedisServer)

func WithLuaFastPathObserver 

func WithLuaFastPathObserver(observer monitoring.LuaFastPathObserver) RedisServerOption

WithLuaFastPathObserver enables per-redis.call() fast-path outcome metrics inside Lua scripts. Used to diagnose fast-path hit ratios for commands like ZRANGEBYSCORE / ZSCORE / HGET.

Resolves per-command counter handles up front so the hot path avoids CounterVec.WithLabelValues on every redis.call().

func WithLuaObserver 

func WithLuaObserver(observer monitoring.LuaScriptObserver) RedisServerOption

WithLuaObserver enables per-phase Lua script metrics (VM exec, Raft commit, retries).

func WithLuaPoolMaxIdle 

func WithLuaPoolMaxIdle(n int) RedisServerOption

WithLuaPoolMaxIdle caps the number of idle *lua.LState instances the Lua VM pool retains between EVALs. The cap controls the steady- state memory floor of the pool (maxIdle * per-state footprint — empirically ~200 KiB) without bounding throughput: get() falls through to a fresh allocation when the pool is empty, and put() drops a state to the GC when the pool is full. n <= 0 is clamped to DefaultLuaPoolMaxIdle, matching newLuaStatePoolWithMaxIdle.

Passing this option overrides the default. The option records the requested cap on the RedisServer; the pool itself is constructed after all options are applied so the recorded cap takes effect.

func WithOnePhaseTxnDedup 

func WithOnePhaseTxnDedup(enabled bool) RedisServerOption

WithOnePhaseTxnDedup enables (or disables) the option-2 one-phase idempotency dedup on list-push and MULTI/EXEC retries (see RedisServer.onePhaseTxnDedup). On by default since the rollout recorded in docs/design/2026_06_10_implemented_redis_onephase_dedup_default_on.md; pass false to opt out from code, or set ELASTICKV_REDIS_ONEPHASE_DEDUP=0 to opt out from the environment. The constructor option trumps the env var.

func WithRedisActiveTimestampTracker 

func WithRedisActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) RedisServerOption

func WithRedisApplyObserver 

func WithRedisApplyObserver(observer *RedisApplyObserver) RedisServerOption

WithRedisApplyObserver makes RedisServer use the same waiter registries that the FSM apply observer signals.

func WithRedisBlockWaitFallback 

func WithRedisBlockWaitFallback(d time.Duration) RedisServerOption

WithRedisBlockWaitFallback overrides the safety-net polling interval for blocking Redis commands. Production should normally use the default and rely on waiter signals for immediate wakeups; tests use a shorter interval to keep package runtime bounded.

func WithRedisCompactor 

func WithRedisCompactor(c *DeltaCompactor) RedisServerOption

WithRedisCompactor wires a DeltaCompactor to the RedisServer so that urgent single-key compaction can be triggered when ErrDeltaScanTruncated is hit.

func WithRedisHeavyCommandSlots 

func WithRedisHeavyCommandSlots(n int) RedisServerOption

WithRedisHeavyCommandSlots overrides the heavy-command limiter size. n <= 0 disables the limiter. Omit the option to use the environment/default selected by newDefaultRedisHeavyCommandLimiter.

func WithRedisLegacyTTLReadFallback 

func WithRedisLegacyTTLReadFallback(enabled bool) RedisServerOption

WithRedisLegacyTTLReadFallback controls whether pre-inline !redis|ttl| entries are still consulted after the anchor-specific TTL probes miss. Leave this enabled during rolling upgrades; disable it only after the background inline TTL migrator has completed for the cluster and legacy bare string keys have been separately ruled out or rewritten.

func WithRedisPerPeerConnectionLimit 

func WithRedisPerPeerConnectionLimit(n int) RedisServerOption

WithRedisPerPeerConnectionLimit overrides the per-peer Redis connection cap. n <= 0 disables the cap.

func WithRedisRequestObserver 

func WithRedisRequestObserver(observer monitoring.RedisRequestObserver) RedisServerOption

WithRedisRequestObserver enables Prometheus-compatible request metrics.

type S3BlobBackfillConfig 

type S3BlobBackfillConfig struct {
	Workers      int
	QueueSize    int
	RatePerPeer  int
	BurstPerPeer int
	ScanInterval time.Duration
	ScanPageSize int
	MaxAttempts  int
	RetryInitial time.Duration
	RetryMax     time.Duration
}

S3BlobBackfillConfig bounds follower catch-up work independently from the Raft apply loop and from user-path proxy-on-miss reads.

func S3BlobBackfillConfigFromEnv 

func S3BlobBackfillConfigFromEnv() (S3BlobBackfillConfig, error)

S3BlobBackfillConfigFromEnv loads the operational bounds for async follower blob fetch. Invalid or non-positive values fail startup instead of silently removing a resource bound.

type S3BlobBackfillObserver 

type S3BlobBackfillObserver interface {
	ObserveS3ChunkBlobBackfillQueueDepth(depth int)
	ObserveS3ChunkBlobBackfillResult(result string)
}

S3BlobBackfillObserver is an optional extension implemented by the metrics observer. Keeping it separate preserves compatibility with small test and embedding observers that only implement the M1 metric surface.

type S3BlobBackfiller 

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

S3BlobBackfiller receives non-blocking FSM apply notifications and performs all reads, peer RPCs, verification, retries, and local writes on bounded background workers.

func NewS3BlobBackfiller 

func NewS3BlobBackfiller(cfg S3BlobBackfillConfig) *S3BlobBackfiller

func (*S3BlobBackfiller) OnApply 

func (b *S3BlobBackfiller) OnApply(op pb.Op, key []byte)

OnApply stays non-blocking because it runs inline on the Raft apply goroutine. A full queue drops the notification; the periodic scanner is the durable recovery path for that case.

func (*S3BlobBackfiller) Start 

func (b *S3BlobBackfiller) Start(ctx context.Context, server *S3Server) error

func (*S3BlobBackfiller) Stop 

func (b *S3BlobBackfiller) Stop()

type S3BlobCluster 

type S3BlobCluster interface {
	S3BlobOffloadCapabilityChecker
	SelfNodeID() string
	ReplicasForChunk(ctx context.Context, chunkKey []byte) ([]S3BlobReplica, error)
	PushChunkBlob(ctx context.Context, replica S3BlobReplica, digest [s3ChunkBlobSHA256Bytes]byte, payload []byte, commitTS uint64) error
	FetchChunkBlob(ctx context.Context, replica S3BlobReplica, digest [s3ChunkBlobSHA256Bytes]byte) ([]byte, error)
	Close() error
}

S3BlobCluster combines capability negotiation and peer-local blob transfer. Implementations return current owning-group membership for every operation; callers calculate durability from that authoritative set.

func NewGRPCS3BlobCluster 

func NewGRPCS3BlobCluster(selfNodeID string, members kv.RaftMembershipCoordinator, adminToken, peerToken string) S3BlobCluster

NewGRPCS3BlobCluster constructs the production peer client. An empty token always fails capability negotiation because S3BlobFetch is peer-only and is never exposed by the insecure Admin mode.

type S3BlobFetchServer 

type S3BlobFetchServer struct {
	pb.UnimplementedS3BlobFetchServer
	// contains filtered or unexported fields
}

S3BlobFetchServer serves local content-addressed S3 chunk blobs. It is the internal peer-to-peer RPC substrate for the chunkref/chunkblob rollout; it deliberately does not enable the public S3 PUT/GET offload path by itself.

func NewS3BlobFetchServer 

func NewS3BlobFetchServer(st store.MVCCStore, observer S3BlobOffloadObserver, opts ...S3BlobFetchServerOption) *S3BlobFetchServer

func (*S3BlobFetchServer) FetchChunkBlob 

func (s *S3BlobFetchServer) FetchChunkBlob(req *pb.FetchChunkBlobRequest, stream pb.S3BlobFetch_FetchChunkBlobServer) error

func (*S3BlobFetchServer) PushChunkBlob 

func (s *S3BlobFetchServer) PushChunkBlob(stream pb.S3BlobFetch_PushChunkBlobServer) error

type S3BlobFetchServerOption 

type S3BlobFetchServerOption func(*S3BlobFetchServer)

func WithS3BlobFetchClock 

func WithS3BlobFetchClock(clock *kv.HLC) S3BlobFetchServerOption

func WithS3BlobFetchPushBlocked 

func WithS3BlobFetchPushBlocked(blocked func() bool) S3BlobFetchServerOption

type S3BlobLocalStoreResolver 

type S3BlobLocalStoreResolver interface {
	LocalStoreForKey(key []byte) (store.MVCCStore, bool)
}

S3BlobLocalStoreResolver exposes this process's store for a chunkblob key. This bypasses leader routing because chunkblob rows are peer-local auxiliary state and never enter Raft.

type S3BlobLocalStoreScanner 

type S3BlobLocalStoreScanner interface {
	LocalStores() []store.MVCCStore
}

S3BlobLocalStoreScanner exposes every local shard store so startup and periodic scans can recover chunkrefs not observed through the live apply hook, including refs installed by snapshot restore.

type S3BlobOffloadCapabilityChecker 

type S3BlobOffloadCapabilityChecker interface {
	AllPeersSupportS3BlobOffload(ctx context.Context) bool
}

S3BlobOffloadCapabilityChecker is the cluster-wide mixed-version guard for admitting an S3 PUT into the offloaded chunkref/chunkblob keyspace.

type S3BlobOffloadObserver 

type S3BlobOffloadObserver interface {
	ObserveS3BlobOffloadDecision(mode, reason string)
	ObserveS3ChunkBlobReplicationDegraded()
	ObserveS3ChunkBlobSHAMismatch()
	ObserveS3ChunkBlobUnrecoverable()
}

S3BlobOffloadObserver records offload rollout and chunkblob durability outcomes. The data path is still fail-closed, so only decision counters are emitted today; the remaining counters are wired for the durable replication PR that will turn the offload path on.

type S3BlobReplica 

type S3BlobReplica struct {
	NodeID   string
	Address  string
	Suffrage string
}

S3BlobReplica is a peer endpoint in the Raft group that owns a chunkblob.

type S3PutAdmissionObserver 

type S3PutAdmissionObserver interface {
	ObserveS3PutAdmissionInflight(bytes int64)
	ObserveS3PutAdmissionRejection(stage, protocol string)
	ObserveS3PutAdmissionWait(stage, protocol string, duration time.Duration)
}

type S3Server 

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

func NewS3Server 

func NewS3Server(listen net.Listener, s3Addr string, st store.MVCCStore, coordinate kv.Coordinator, leaderS3 map[string]string, opts ...S3ServerOption) *S3Server

func (*S3Server) AdminCreateBucket 

func (s *S3Server) AdminCreateBucket(ctx context.Context, principal AdminPrincipal, name, acl string) (*AdminBucketSummary, error)

AdminCreateBucket creates a bucket on behalf of the admin dashboard. The principal MUST be re-validated by the caller (the admin HTTP handler does this against the live RoleStore); this method enforces the authorisation invariant a second time so a follower-forwarded call cannot smuggle a read-only principal past the check on the leader side (Section 3.2 "認可の真実は常に adapter 側").

The transaction is atomic: bucket meta + generation + ACL all land in a single OperationGroup, mirroring the SigV4 createBucket path. On success returns the freshly-stored summary; on conflict returns ErrAdminBucketAlreadyExists; on a non-leader / non-full-role / bad input returns the corresponding sentinel.

func (*S3Server) AdminDeleteBucket 

func (s *S3Server) AdminDeleteBucket(ctx context.Context, principal AdminPrincipal, name string) error

AdminDeleteBucket removes a bucket if it is empty. Same authorisation contract as the other admin write methods. The bucket-must-be-empty rule mirrors the SigV4 deleteBucket path — the dashboard cannot force a recursive delete, by design.

The dispatch happens in two phases because the production coordinator (kv/sharded_coordinator.go: dispatchDelPrefixBroadcast) rejects DEL_PREFIX inside a transaction and rejects DEL_PREFIX mixed with Del or Put in the same OperationGroup: 

Phase 1: Del BucketMetaKey in a txn (OCC-protected against
         a concurrent AdminCreateBucket landing between our
         readTS and commitTS).
Phase 2: DEL_PREFIX over every per-bucket key family in a
         non-txn broadcast — the safety net that sweeps
         orphans left by any PutObject that committed
         chunks/manifest between the empty-probe and the
         Phase-1 commit. See design doc
         2026_04_28_implemented_admin_delete_bucket_safety_net.md
         §6.2 for the original single-OperationGroup design
         and the dispatch-shape rejection that forced the
         two-phase split.

Phase 2 is best-effort: a Phase-2 failure leaves the bucket meta already deleted (Phase 1 succeeded) but per-bucket prefixes possibly still containing orphans. That state is no worse than the pre-fix behaviour on main and recovers on operator-driven re-cleanup. We log a warning rather than propagate the error so the operator-visible delete reports success — the bucket really is gone from the API surface, and a retry would 404 because loadBucketMetaAt no longer finds the meta.

BucketGenerationKey is intentionally NOT deleted. Re-creating the bucket bumps the generation; orphan blobs that escaped this delete (e.g. on an older generation) stay isolated under the old generation prefix and never surface in the new bucket. Pinned by TestS3Server_AdminDeleteBucket_BucketGenerationKeySurvives.

The contract change for clients: a PutObject that returned 200 OK during the race window can have its data swept by the concurrent delete. Operators are advised to pause writes before AdminDeleteBucket; the alternative (orphan objects that no API can enumerate or remove) is strictly worse.

The same shape is mirrored on the SigV4 path (adapter/s3.go:deleteBucket) so both delete entrypoints share the same race-window guarantees.

func (*S3Server) AdminDeleteObject 

func (s *S3Server) AdminDeleteObject(ctx context.Context, principal AdminPrincipal, bucket, key string) error

AdminDeleteObject removes one object from a bucket. Write role required. Idempotent: a missing object returns nil (matches the SigV4 deleteObject path and AWS S3 semantics).

Sentinels:

  • ErrAdminForbidden — principal lacks write role
  • ErrAdminNotLeader — follower
  • ErrAdminBucketNotFound — bucket absent
  • ErrAdminInvalidObjectKey — empty key

func (*S3Server) AdminDescribeBucket 

func (s *S3Server) AdminDescribeBucket(ctx context.Context, name string) (*AdminBucketSummary, bool, error)

AdminDescribeBucket returns the bucket-level snapshot for name. The triple (result, present, error) lets admin callers distinguish a genuine "not found" from a storage error without sniffing sentinels — when the bucket is missing the function returns (nil, false, nil), mirroring AdminDescribeTable's contract on the Dynamo side.

Like AdminListBuckets this is a read-only path that bypasses SigV4. The HTTP admin handler enforces session + CSRF + role at the boundary; the adapter trusts the caller for authentication (Section 3.2's exception for read-only paths).

func (*S3Server) AdminGetObject 

func (s *S3Server) AdminGetObject(ctx context.Context, principal AdminPrincipal, bucket, key string) (io.ReadCloser, AdminObject, error)

AdminGetObject fetches one object's body + metadata. Read role required. Returns (body, meta, nil) on success; the caller MUST Close the body to release the read-tracker pin.

Sentinels:

  • ErrAdminForbidden — principal lacks read role
  • ErrAdminNotLeader — follower
  • ErrAdminBucketNotFound — bucket absent
  • ErrAdminObjectNotFound — object absent
  • ErrAdminInvalidObjectKey — empty key
  • ErrAdminInvalidBucketName — malformed bucket name

func (*S3Server) AdminListBuckets 

func (s *S3Server) AdminListBuckets(ctx context.Context) ([]AdminBucketSummary, error)

AdminListBuckets returns every S3-style bucket this server knows about, in lexicographic order (the metadata-prefix scan natural ordering). Intended for the in-process admin listener as the SigV4-free counterpart to the listBuckets HTTP handler.

Unlike the SigV4 path (which intentionally caps each call at s3MaxKeys = 1000 because the AWS API is page-based), the admin dashboard's pagination is implemented at the handler layer, which expects this method to return the full set. We loop the per-page ScanAt until the metadata prefix is exhausted — same pattern as scanAllByPrefixAt on the Dynamo side (Codex P1 + Claude Issue 1 on PR #658).

Returns an empty slice (not nil) when no buckets exist so JSON callers see `[]` instead of `null`.

func (*S3Server) AdminListObjects 

func (s *S3Server) AdminListObjects(ctx context.Context, principal AdminPrincipal, bucket string, opts AdminListObjectsOptions) (AdminObjectListing, error)

AdminListObjects returns a bounded page of objects under a prefix. Read role required. Delimiter collapses pseudo-folders into CommonPrefixes the same way SigV4 ListObjectsV2 does.

Sentinels:

  • ErrAdminForbidden — principal lacks read role
  • ErrAdminNotLeader — follower
  • ErrAdminBucketNotFound — bucket absent
  • ErrAdminInvalidBucketName — malformed bucket name argument
  • ErrAdminInvalidContinuationToken — token references a different bucket / generation / prefix / delimiter than the current request, OR the token's readTS has been MVCC-GC'd past (store.ErrReadTSCompacted on the underlying scan)

A structurally invalid / non-base64-decodable continuation token surfaces as a wrapped json/decode error from decodeS3ContinuationToken, not as a sentinel — the bridge maps that to a generic 400 with the decode-error text preserved.

Mirrors the SigV4 listObjectsV2 precedent at s3.go (same scan- and-collapse pipeline with one error-return per stage; carries the same nolint set). Splitting further would obscure the pagination cursor lifecycle.

func (*S3Server) AdminPutBucketAcl 

func (s *S3Server) AdminPutBucketAcl(ctx context.Context, principal AdminPrincipal, name, acl string) error

AdminPutBucketAcl swaps the canned ACL on an existing bucket. Same authorisation contract as AdminCreateBucket. Mutates only the meta.Acl field; generation is preserved so existing object references stay valid.

func (*S3Server) AdminPutObject 

func (s *S3Server) AdminPutObject(ctx context.Context, principal AdminPrincipal, bucket, key string, body io.Reader, contentType string) error

AdminPutObject creates-or-replaces one object via a streaming upload. Write role required. Caps body size at adminS3UploadMaxBytes (100 MiB) per design §3.3.3; larger objects must use the public SigV4 endpoint with the SDK's multipart path.

contentType is stored verbatim on the manifest (defaults to application/octet-stream when empty). body is consumed exactly once; the caller owns its lifecycle (close, etc.) — AdminPutObject does NOT close the reader.

Sentinels:

  • ErrAdminForbidden — principal lacks write role
  • ErrAdminNotLeader — follower
  • ErrAdminBucketNotFound — bucket absent
  • ErrAdminInvalidObjectKey — empty key
  • ErrAdminInvalidBucketName — malformed bucket name
  • ErrAdminUploadTooLarge — body exceeds adminS3UploadMaxBytes

func (*S3Server) Run 

func (s *S3Server) Run() error

func (*S3Server) SetListener 

func (s *S3Server) SetListener(listen net.Listener)

SetListener installs the listener Run serves from. Startup wiring uses this to build admin-forward sources before binding the public socket.

func (*S3Server) StartBlobBackfill 

func (s *S3Server) StartBlobBackfill(ctx context.Context) error

func (*S3Server) Stop 

func (s *S3Server) Stop()

type S3ServerOption 

type S3ServerOption func(*S3Server)

func WithS3ActiveTimestampTracker 

func WithS3ActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) S3ServerOption

func WithS3BlobBackfiller 

func WithS3BlobBackfiller(backfiller *S3BlobBackfiller) S3ServerOption

func WithS3BlobCluster 

func WithS3BlobCluster(cluster S3BlobCluster) S3ServerOption

func WithS3BlobLocalStoreResolver 

func WithS3BlobLocalStoreResolver(resolver S3BlobLocalStoreResolver) S3ServerOption

func WithS3BlobMinReplicas 

func WithS3BlobMinReplicas(minReplicas int) S3ServerOption

func WithS3BlobOffloadCapabilityChecker 

func WithS3BlobOffloadCapabilityChecker(checker S3BlobOffloadCapabilityChecker) S3ServerOption

func WithS3BlobOffloadEnabled 

func WithS3BlobOffloadEnabled(enabled bool) S3ServerOption

func WithS3BlobOffloadObserver 

func WithS3BlobOffloadObserver(observer S3BlobOffloadObserver) S3ServerOption

func WithS3BlobPushBlocked 

func WithS3BlobPushBlocked(blocked func() bool) S3ServerOption

func WithS3PutAdmissionObserver 

func WithS3PutAdmissionObserver(observer S3PutAdmissionObserver) S3ServerOption

func WithS3Region 

func WithS3Region(region string) S3ServerOption

func WithS3StaticCredentials 

func WithS3StaticCredentials(creds map[string]string) S3ServerOption

type SQSPartitionObserver 

type SQSPartitionObserver interface {
	ObservePartitionMessage(queue string, partition uint32, action string)
}

SQSPartitionObserver is the metrics-package interface (monitoring.SQSPartitionObserver) re-declared here so the adapter does not import monitoring at the package boundary — matches the existing observer pattern for DynamoDB / Redis.

type SQSPartitionResolver 

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

SQSPartitionResolver maps a partitioned-SQS key to the operator- chosen Raft group for the (queue, partition) tuple. Implements kv.PartitionResolver via duck typing — see the integration in main.go where the resolver is installed on ShardedCoordinator.

The byte-range engine cannot route partitioned queues because adding per-partition routes would break its non-overlapping-cover invariant (a partition route for partition K of one queue would leave a gap for legacy keys that fall lexicographically between partitions K and K+1). The resolver-first dispatch path avoids this — it answers only for keys that match a partitioned family prefix and otherwise lets the engine handle dispatch.

func NewSQSPartitionResolver 

func NewSQSPartitionResolver(routes map[string][]uint64) *SQSPartitionResolver

NewSQSPartitionResolver builds a resolver from the operator- supplied partition map. routes[queue][k] is the Raft group ID that owns partition k of queue, with len(routes[queue]) equal to the queue's PartitionCount.

Returns nil when routes is empty so callers can keep the resolver out of the request path entirely on a non-partitioned cluster (kv.ShardRouter.WithPartitionResolver(nil) is a documented no-op).

The constructor takes a defensive copy so a later caller mutation to the input map does not leak into the resolver's view at runtime.

func (*SQSPartitionResolver) RecognisesPartitionedKey 

func (r *SQSPartitionResolver) RecognisesPartitionedKey(key []byte) bool

RecognisesPartitionedKey reports whether key has the structural shape of a partitioned-SQS key — i.e. starts with one of the partitioned family prefixes. The check is PREFIX-ONLY, not a full parse: a key with a partitioned prefix followed by a malformed queue / partition segment still answers true, so the router fails closed via kv.PartitionResolver semantics instead of falling through to the engine and silently routing to the SQS catalog default group via routeKey's !sqs|route|global collapse (round 5 review nit on PR #715).

A nil receiver returns false so kv.ShardRouter's typed-nil case (ResolveGroup(nil) == (0, false)) pairs with an honest "I don't recognise anything" answer instead of falsely claiming a shape.

func (*SQSPartitionResolver) ResolveGroup 

func (r *SQSPartitionResolver) ResolveGroup(key []byte) (uint64, bool)

ResolveGroup decodes the (queue, partition) embedded in a partitioned-SQS key and returns the operator-chosen Raft group.

Returns (0, false) for any key that does not match a partitioned family prefix (legacy SQS, KV, S3, DynamoDB, queue-meta records, …) so kv.ShardRouter falls through to its byte-range engine for default routing.

Returns (0, false) for a partitioned-shaped key whose queue is not in the routes map or whose partition index is beyond len(routes[queue]). The router pairs this with RecognisesPartitionedKey to fail closed instead of falling through — silently routing through the engine's !sqs|route|global default would mis-route HT-FIFO traffic during partition-map drift (codex P1 round 2 on PR #715).

func (*SQSPartitionResolver) RoutedPartitionCount 

func (r *SQSPartitionResolver) RoutedPartitionCount(queueName string) int

RoutedPartitionCount returns the number of partition routes configured for queueName, or 0 if the queue is not in the routing map. Used by the CreateQueue capability gate (validateHTFIFOCapability) to verify that EVERY partition of a requested partitioned queue is routable BEFORE the create commits — without this, a queue could land with PartitionCount=N but only K<N routes, and SendMessage on the missing partitions would fail closed at the router with "no route for key" (Codex P1 review on PR #734).

A nil receiver returns 0 so the gate's "resolver==nil → skip the coverage check" branch kicks in cleanly: a single-shard / no---sqsFifoPartitionMap deployment has no per-partition routing to verify, and partitioned keys fall through to the engine's default group.

type SQSQueueDepth 

type SQSQueueDepth struct {
	Queue      string
	Visible    int64
	NotVisible int64
	Delayed    int64
}

SQSQueueDepth is one queue's depth-attribute snapshot, the unit the SQSServer hands to monitoring.SQSObserver on each tick. The fields mirror sqsApproxCounters byte-for-byte and the public AdminQueueCounters JSON shape — operators see consistent numbers in dashboards and the admin SPA.

type SQSServer 

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

func NewSQSServer 

func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordinator, opts ...SQSServerOption) *SQSServer

func (*SQSServer) AdminDeleteQueue 

func (s *SQSServer) AdminDeleteQueue(ctx context.Context, principal AdminPrincipal, name string) error

AdminDeleteQueue is the SigV4-bypass counterpart to deleteQueue. Returns the same sentinel errors as AdminCreateTable on the Dynamo side: ErrAdminForbidden on a read-only principal, ErrAdminNotLeader on a follower, ErrAdminSQSNotFound when the queue is absent.

func (*SQSServer) AdminDescribeQueue 

func (s *SQSServer) AdminDescribeQueue(ctx context.Context, name string, opts ...AdminDescribeQueueOptions) (*AdminQueueSummary, bool, error)

AdminDescribeQueue returns a snapshot of name's metadata plus the approximate counters. The triple (result, present, error) lets admin callers distinguish a missing queue from a storage error without sniffing sentinels.

Like AdminDescribeTable on the Dynamo side, this entrypoint runs on either the leader or a follower (read-only); the counter scan uses a fresh nextTxnReadTS so the result is consistent with what SigV4 GetQueueAttributes would have returned at the same instant.

Optional AdminDescribeQueueOptions toggle additional lookups (currently DLQ source enumeration). Existing callers that pass no options retain the cheap path; Phase 4 wiring opts in when the SPA needs the DLQ relationship.

func (*SQSServer) AdminListQueues 

func (s *SQSServer) AdminListQueues(ctx context.Context) ([]string, error)

AdminListQueues returns every queue name this server knows about, in the lexicographic order the queue catalog index produces. Read path; runs on follower or leader and uses the same scanQueueNames helper the SigV4 ListQueues handler does.

func (*SQSServer) AdminPeekQueue 

func (s *SQSServer) AdminPeekQueue(
	ctx context.Context,
	principal AdminPrincipal,
	name string,
	opts AdminPeekMessageOptions,
) ([]AdminPeekedMessage, string, error)

AdminPeekQueue returns a non-destructive sample of currently-visible messages in name. Receive counts are NOT incremented and visibility timers are NOT started — peek is a pure read over the leader's visibility index. Returns the rows plus a continuation cursor that the caller passes back as opts.Cursor to fetch the next page (empty when the walk is complete).

Sentinel errors:

  • ErrAdminForbidden — peek requires read role; nil principal is denied
  • ErrAdminNotLeader — peek runs on the leader (the visibility index is leader-only-written; a follower read would race the leader's apply)
  • ErrAdminSQSNotFound — queue absent
  • ErrAdminSQSValidation — empty / malformed name, malformed / oversized / stale-generation cursor

func (*SQSServer) AdminPurgeQueue 

func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipal, name string) (AdminPurgeResult, error)

AdminPurgeQueue is the SigV4-bypass counterpart to purgeQueue. Bumps the queue's generation so every message under the old generation becomes unreachable, leaving the meta record (name, ARN, RedrivePolicy, tags, attributes) in place. The reaper eventually deletes the orphaned message keys via the existing tombstone path.

Returns the captured generation pair so the admin handler's audit line can record the value that actually landed; a separate loadQueueMetaAt call would race a concurrent purge in the 60-second window. The 60-second rate limit lives on the meta record itself so the SigV4 and admin paths interlock uniformly.

Sentinel errors:

  • ErrAdminForbidden — read-only principal
  • ErrAdminNotLeader — follower
  • ErrAdminSQSNotFound — queue absent
  • *PurgeInProgressError — last purge < 60 s ago; errors.Is matches ErrAdminSQSPurgeInProgress, RetryAfter carries the duration.
  • ErrAdminSQSValidation — empty / whitespace name

func (*SQSServer) AdminSetQueueAttributes 

func (s *SQSServer) AdminSetQueueAttributes(ctx context.Context, principal AdminPrincipal, name string, attrs map[string]string) error

AdminSetQueueAttributes is the SigV4-bypass counterpart to SetQueueAttributes. It is intentionally generic rather than DLQ-specific so the admin SPA can edit RedrivePolicy and RedriveAllowPolicy through the same validator the public SQS API uses.

func (*SQSServer) Run 

func (s *SQSServer) Run() error

func (*SQSServer) SnapshotQueueDepths 

func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) ([]SQSQueueDepth, bool)

SnapshotQueueDepths satisfies monitoring.SQSDepthSource. The observer Start loop calls this on every tick.

Returns:

  • (snaps, true) — leader, scrape OK. snaps is non-nil, even when there are zero queues. Observer writes snaps to the gauges and diffs against the previous tick (forgetting any queue that disappeared from this snapshot).
  • (nil, true) — this node is a follower (leader-only emission keeps gauges consistent with AdminListQueues / AdminDescribeQueue at the same instant — follower scans would race the leader's writes). Empty-but-OK so the observer ForgetQueue's any gauges this node was emitting before stepping down.
  • (nil, false) — leader, but scrape failed (transient catalog-read error or ctx cancel mid-scan). Tells the observer to skip this tick: leave existing gauges in place rather than wiping every depth series — a single failed scrape would otherwise dashboard-render as a false "all queues drained" event until the next successful tick.

Per-queue scan errors (loadQueueMetaAt / scanApproxCounters) remain handled in-line by snapshotOneQueueDepth: the offending queue is dropped from this tick's snapshot but ok stays true, so the observer ForgetQueue's just that one queue's gauges. Only a top-level scanQueueNames failure (which would silently turn into "no queues anywhere") flips ok to false.

func (*SQSServer) Stop 

func (s *SQSServer) Stop()

type SQSServerOption 

type SQSServerOption func(*SQSServer)

func WithSQSLeaderMap 

func WithSQSLeaderMap(m map[string]string) SQSServerOption

WithSQSLeaderMap configures the Raft-address-to-SQS-address mapping used to forward requests from followers to the current leader. Format mirrors WithDynamoDBLeaderMap / WithS3LeaderMap.

func WithSQSPartitionObserver 

func WithSQSPartitionObserver(o SQSPartitionObserver) SQSServerOption

WithSQSPartitionObserver installs the elastickv_sqs_partition_messages_total counter observer on the SQS server. Pass nil (the default) on non-monitored test fixtures; the partitioned send / receive / delete paths then observe via a nil interface and the metric stays at zero. The monitoring registry's SQSPartitionObserver() returns the concrete implementation in production.

func WithSQSPartitionResolver 

func WithSQSPartitionResolver(r *SQSPartitionResolver) SQSServerOption

WithSQSPartitionResolver installs the cluster's partition resolver on the SQS server so the CreateQueue capability gate (validateHTFIFOCapability) can verify routing coverage before admitting a partitioned create. Pass nil (the default) on single-shard / no---sqsFifoPartitionMap deployments — the gate then skips the coverage check.

Callers must ensure the resolver passed here matches the one installed on the kv coordinator via WithPartitionResolver, otherwise the gate would admit a queue that the coordinator then fails to route. main.go builds the resolver once and hands the same pointer to both consumers.

func WithSQSRegion 

func WithSQSRegion(region string) SQSServerOption

WithSQSRegion configures the signing region the adapter expects inside the Credential scope. Empty values retain the previous setting.

func WithSQSStaticCredentials 

func WithSQSStaticCredentials(creds map[string]string) SQSServerOption

WithSQSStaticCredentials supplies the access-key → secret map the adapter will accept. Passing an empty map disables authorization entirely (open endpoint), matching the S3 adapter's behavior for unit-test friendliness.

func WithSQSThrottleObserver 

func WithSQSThrottleObserver(o SQSThrottleObserver) SQSServerOption

WithSQSThrottleObserver installs the elastickv_sqs_throttled_requests_total and elastickv_sqs_throttle_tokens_remaining observer on the SQS server. Pass nil (the default) on non-monitored fixtures.

type SQSThrottleObserver 

type SQSThrottleObserver interface {
	ObserveThrottleDecision(queue string, action string, tokensRemaining float64, throttled bool)
	ForgetThrottleAction(queue string, action string)
	SyncThrottleActions(queue string, enabledActions []string)
}

SQSThrottleObserver is the metrics-package interface (monitoring.SQSThrottleObserver) re-declared here so the adapter does not import monitoring at the package boundary.

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.