runtime

Documentation

Overview

Package runtime provides the fail-closed delivery loop used by a real AntiFlock endpoint agent.

Index

• func LoadFileSigner(nodeID, path string, clock func() time.Time) (collectors.EventSigner, error)
• func LoadNodeCertificate(certificatePath, seedPath string) (tls.Certificate, error)
• type Collector
• type CollectorFunc
  • func (function CollectorFunc) Collect(ctx context.Context) (*collectors.Collection, error)
• type Loop
  • func NewLoop(config LoopConfig) (*Loop, error)
  • func (loop *Loop) Run(ctx context.Context) error
  • func (loop *Loop) RunOnce(ctx context.Context) error
• type LoopConfig
• type Queue
  • func OpenQueue(directory, nodeID string) (*Queue, error)
  • func (queue *Queue) Acknowledge(ctx context.Context, events []*antiflockv1.EventEnvelope) error
  • func (queue *Queue) Batch(ctx context.Context, maximum int) ([]*antiflockv1.EventEnvelope, error)
  • func (queue *Queue) Close() error
  • func (queue *Queue) Enqueue(ctx context.Context, event *antiflockv1.EventEnvelope, ...) error
  • func (queue *Queue) EnqueueBatch(ctx context.Context, entries []QueueEntry) error
  • func (queue *Queue) NextSequence(ctx context.Context) (uint64, error)
• type QueueEntry
• type QueueStatus
  • func InspectQueue(directory, nodeID string) (QueueStatus, error)
• type Submitter

Functions

func LoadFileSigner

func LoadFileSigner(nodeID, path string, clock func() time.Time) (collectors.EventSigner, error)

LoadFileSigner loads the Ed25519 seed created during enrollment. The seed file must be private and regular; it is never copied into the queue.

func LoadNodeCertificate

func LoadNodeCertificate(certificatePath, seedPath string) (tls.Certificate, error)

LoadNodeCertificate binds the approved mTLS certificate to the same Ed25519 seed used for source-event signing. It avoids a second private-key copy.

Types

type Collector

type Collector interface {
	Collect(context.Context) (*collectors.Collection, error)
}

Collector is the narrow, read-only observation source consumed by Loop.

type CollectorFunc

type CollectorFunc func(context.Context) (*collectors.Collection, error)

CollectorFunc adapts a verified read-only collection function to the loop.

func (CollectorFunc) Collect

func (function CollectorFunc) Collect(ctx context.Context) (*collectors.Collection, error)

type Loop

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

Loop performs continuous collection without host mutation. Collection failure leaves already queued telemetry untouched; submission failure leaves the exact signed batch on disk for the next attempt.

func NewLoop

func NewLoop(config LoopConfig) (*Loop, error)

func (*Loop) Run

func (loop *Loop) Run(ctx context.Context) error

Run executes an immediate collection, then repeats at the configured interval. A collection or delivery failure leaves the durable queue intact and is retried on the next interval; --once callers use RunOnce directly when they need the error returned to their supervisor.

func (*Loop) RunOnce

func (loop *Loop) RunOnce(ctx context.Context) error

type LoopConfig

type LoopConfig struct {
	DeploymentID string
	NodeID       string
	BootID       string
	Interval     time.Duration
	Collector    Collector
	Queue        *Queue
	Signer       collectors.EventSigner
	Submitter    Submitter
	Clock        func() time.Time
}

type Queue

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

Queue is a bounded, private on-disk write-ahead queue. It stores signed protobuf envelopes verbatim; retries never re-sign or regenerate them.

func OpenQueue

func OpenQueue(directory, nodeID string) (*Queue, error)

func (*Queue) Acknowledge

func (queue *Queue) Acknowledge(ctx context.Context, events []*antiflockv1.EventEnvelope) error

Acknowledge removes exactly the events sent in a rejection-free Core batch. The Core's contiguous cursor is intentionally not used as a deletion signal: source sequence gaps are legal, while the acknowledgement proves durability.

func (*Queue) Batch

func (queue *Queue) Batch(ctx context.Context, maximum int) ([]*antiflockv1.EventEnvelope, error)

Batch returns events in their source sequence order. It does not mutate the queue; callers must call Acknowledge only after the Core acknowledgement.

func (*Queue) Close

func (queue *Queue) Close() error

Close releases the process-wide writer lock. A caller must not continue to use the queue after Close; the agent defers it for orderly service shutdown.

func (*Queue) Enqueue

func (queue *Queue) Enqueue(ctx context.Context, event *antiflockv1.EventEnvelope, priority collectors.QueuePriority) error

func (*Queue) EnqueueBatch

func (queue *Queue) EnqueueBatch(ctx context.Context, entries []QueueEntry) error

EnqueueBatch persists a complete collection cycle in one queue state update. It never leaves a partial cycle behind when capacity or disk persistence fails.

func (*Queue) NextSequence

func (queue *Queue) NextSequence(ctx context.Context) (uint64, error)

type QueueEntry

type QueueEntry struct {
	Event    *antiflockv1.EventEnvelope
	Priority collectors.QueuePriority
}

QueueEntry is one already-signed envelope prepared for atomic queue admission.

type QueueStatus

type QueueStatus struct {
	SchemaVersion  string `json:"schemaVersion"`
	NodeID         string `json:"nodeId"`
	LastSequence   uint64 `json:"lastSequence"`
	RetainedEvents int    `json:"retainedEvents"`
	MaximumEvents  int    `json:"maximumEvents"`
}

QueueStatus is safe local operator metadata. It contains neither event payloads nor signing material and may be inspected while another process owns the queue writer lock.

func InspectQueue

func InspectQueue(directory, nodeID string) (QueueStatus, error)

InspectQueue reads a complete atomically-installed queue state without acquiring its writer lock. It never creates, changes, or acknowledges data. An active writer only exposes either the prior complete file or its complete replacement, never a staged temporary file.

type Submitter

type Submitter interface {
	Submit(context.Context, *antiflockv1.SubmitEventBatchRequest) (*antiflockv1.EventBatchAck, error)
}

Submitter is satisfied by ingest.Client. It receives only already-signed envelopes that have first crossed the durable Queue boundary.