Documentation
¶
Overview ¶
Package infra holds the NATS handles shared by every packtrail component — connection, JetStream context, resource names, partition count — and the provisioning of the streams, buckets and object stores they use.
Index ¶
- Constants
- Variables
- func DefaultBlobThreshold(nc *nats.Conn) int
- func Retry(ctx context.Context, attempts int, fn func() error) error
- type Infra
- func (i *Infra) Attach(ctx context.Context) error
- func (i *Infra) CheckServer() error
- func (i *Infra) CmdSubject(execID string) string
- func (i *Infra) EraseKey(ctx context.Context, bucket, key string) error
- func (i *Infra) EraseObject(ctx context.Context, bucket, name string) error
- func (i *Infra) EventSubject(execID string) string
- func (i *Infra) KV(ctx context.Context, bucket string) (jetstream.KeyValue, error)
- func (i *Infra) Object(ctx context.Context, bucket string) (jetstream.ObjectStore, error)
- func (i *Infra) Partition(execID string) int
- func (i *Infra) Provision(ctx context.Context, partitions int) error
Constants ¶
const ( DefaultReadTimeout = 5 * time.Second DefaultBlobTimeout = 2 * time.Minute )
Default timeouts (all configurable on the engine, client and worker).
const DedupWindow = dedupWindow
DedupWindow is how long the command and work streams remember a message id. Ids are derived from the execution id, so an execution id must not be used again before its last messages have left the window: a deleted execution is kept at least this long after it finished.
const DefaultPartitions = 64
DefaultPartitions is the partition count of a new deployment. The count is permanent, so the default leaves room to add engines (G5-08).
const MetaPartitions = "packtrail.partitions"
MetaPartitions is the command-stream metadata key holding the partition count: every process (engine, client, SDK) reads it from there, so all agree on where an execution's commands go.
const MetaProtocol = "packtrail.protocol"
MetaProtocol is the command-stream metadata key holding the wire protocol version.
const ProtocolVersion = "1"
ProtocolVersion is the version of the wire protocol.
Variables ¶
var ErrNotProvisioned = apierr.ErrNotProvisioned
ErrNotProvisioned is returned by Attach when the deployment does not exist.
Functions ¶
func DefaultBlobThreshold ¶
DefaultBlobThreshold derives the claim-check threshold from the server's max_payload minus a margin for headers and subject.
Types ¶
type Infra ¶
type Infra struct {
NC *nats.Conn
JS jetstream.JetStream
Names names.Names
Partitions int
Logger *slog.Logger
// BlobThreshold is the body size above which a message body is moved to
// the blobs object store (claim-check).
BlobThreshold int
// AckWait and PullExpiry configure the engine and dispatcher consumers.
AckWait time.Duration
PullExpiry time.Duration
// ReadTimeout bounds one batched read of an event log.
ReadTimeout time.Duration
// BlobTimeout bounds one blob transfer when the caller sets no deadline.
BlobTimeout time.Duration
// Replicas is the replication factor of every stream, bucket and object
// store (0 = 1). Use 3 in production: the events stream is the source of
// truth.
Replicas int
// contains filtered or unexported fields
}
Infra bundles the shared NATS handles.
func New ¶
New returns an Infra for an existing connection. Partitions and the blob threshold are resolved by Attach (from the deployed command stream) or Provision.
func (*Infra) CheckServer ¶
CheckServer fails fast when the connected server is too old for packtrail (backlog: "require NATS >= 2.12 and fail at startup").
func (*Infra) CmdSubject ¶
CmdSubject returns the command subject of an execution.
func (*Infra) EraseKey ¶ added in v1.0.1
EraseKey removes a key of a KV bucket without leaving a delete marker. Erasing a missing key is a no-op.
func (*Infra) EraseObject ¶ added in v1.0.1
EraseObject removes an object of an object store without leaving a delete marker. Erasing a missing object is a no-op.
func (*Infra) EventSubject ¶
EventSubject returns the event subject of an execution.
Source Files