Documentation
¶
Overview ¶
Package runnerspool stores proxy decisions until the authenticated runner stream acknowledges their durable org ingestion. The proxy and daemon use the same private directory, serialized by a process-shared file lock.
Index ¶
- Constants
- Variables
- type Record
- type Spool
- func (s *Spool) AckOverflow(ctx context.Context, row Record) error
- func (s *Spool) After(ctx context.Context, seq int64, limit int) (rows []Record, err error)
- func (s *Spool) Append(ctx context.Context, decision egress.DecisionLog) error
- func (s *Spool) AppendAdmitted(ctx context.Context, decision egress.DecisionLog) error
- func (s *Spool) CanRecord(ctx context.Context) error
- func (s *Spool) Close() error
- func (s *Spool) Get(ctx context.Context) (pos ackcursor.Pos, err error)
- func (s *Spool) Overflow(ctx context.Context) (row *Record, err error)
- func (s *Spool) Receipt(ctx context.Context) (out types.RunnerSpoolReceipt, err error)
- func (s *Spool) Refuse(ctx context.Context) error
- func (s *Spool) Replay(ctx context.Context, send func(context.Context, []Record) error) error
- func (s *Spool) Seal(ctx context.Context, expected types.RunnerSpoolReceipt) (sealed bool, err error)
- func (s *Spool) Set(ctx context.Context, pos ackcursor.Pos) error
- func (s *Spool) Stats(ctx context.Context) (depth int, bytes int64, drops uint64, err error)
Constants ¶
const ( // ReplayBatch bounds the uncommitted local ACK window retained by org receipts. ReplayBatch = 64 MaxRecords = 50000 MaxBytes = 32 << 20 // MaxRecordBytes leaves the record envelope below the wire control bound. MaxRecordBytes = (1 << 20) - 4096 )
Variables ¶
var ErrFull = errors.New("runner decision spool is full")
var ErrSealed = errors.New("runner spool is durably sealed")
ErrSealed refuses new evidence production after exact settled retirement.
Functions ¶
This section is empty.
Types ¶
type Record ¶
type Record struct {
SpoolID uuid.UUID `json:"spool_id"`
Seq int64 `json:"seq"`
Hash string `json:"hash"`
Decision egress.DecisionLog `json:"decision"`
Count uint64 `json:"count,omitempty"`
DropEnd uint64 `json:"drop_end,omitempty"`
}
Record is one stable, replayable decision. Seq zero is an overflow summary, whose durable identity is its hash; ordinary records have positive sequences.
func (Record) Marshal ¶
Marshal is the bounded EVENT.spool payload; no run token authenticates replay.
type Spool ¶
type Spool struct {
// contains filtered or unexported fields
}
Spool is safe across goroutines and independently opened proxy/daemon handles.
func Open ¶
Open refuses unsafe or corrupt existing state. A missing directory starts a new private stream; an existing stream never silently resets its cursor.
func (*Spool) AckOverflow ¶
AckOverflow records only the frozen interval, retaining later refusals.
func (*Spool) After ¶
After implements ackcursor.Source. The acknowledged row's identity lives in the manifest, so reclaimed payloads do not make ackcursor infer a reset.
func (*Spool) Append ¶
Append commits before returning. An unreserved producer must refuse on any error; the proxy separately reserves capacity before forwarding requests.
func (*Spool) AppendAdmitted ¶
AppendAdmitted finishes a request whose proxy already reserved capacity. Overflow refuses new requests, never discards this previously admitted tail.
func (*Spool) CanRecord ¶
CanRecord checks durable overflow state. Concurrent producers must also reserve capacity; checking alone cannot guarantee space for a later append.
func (*Spool) Overflow ¶
Overflow freezes one cumulative refused-attempt interval before transmission. A crash or new refusals cannot change the identity/count of an in-flight row.
func (*Spool) Replay ¶
Replay uses the shared acknowledgement cursor. send gets each batch in seq order and must return nil only after durable checked org ingestion of every row in it (H2 Peer.EventsAck), not after a write.
Replay ends once it has delivered the head (rows and refused attempts) seen when it started. Rows a live proxy appends meanwhile wait for the next call, so a producer faster than the drain cannot keep Replay running.
func (*Spool) Seal ¶
func (s *Spool) Seal(ctx context.Context, expected types.RunnerSpoolReceipt) (sealed bool, err error)
Seal requires the proxy's lifetime writer lock and an exact fully accepted current head. A still-running producer or unsettled evidence defers sealing.
Source Files