Documentation
¶
Overview ¶
Package audit records, reads, and exports immutable AgentSQL audit events. Database access remains behind the store repository interfaces.
Index ¶
- Constants
- Variables
- func NewBatchError(kind BatchFailureKind, retryable bool, cause error) error
- type BatchBackend
- type BatchError
- type BatchFailureKind
- type GroupCommitSink
- type GroupCommitStats
- type Option
- func SQLiteGroupCommitOptions() []Option
- func WithAttemptTimeout(value time.Duration) Option
- func WithCanonicalDigest(value func(model.AuditLog) (string, error)) Option
- func WithDrainTimeout(value time.Duration) Option
- func WithEnqueueWaitTimeout(value time.Duration) Option
- func WithFlusherExitedHook(value func(chainID string)) Option
- func WithLinger(value time.Duration) Option
- func WithMaxBatch(value int) Option
- func WithMaxEventBytes(value int) Option
- func WithMaxWaiters(value int) Option
- func WithQueueCapacity(value int) Option
- func WithQueueMaxBytes(value int64) Option
- func WithRootTimeout(value time.Duration) Option
- type Reader
- type Recorder
- type Service
- func (service *Service) ExportJSONL(ctx context.Context, filter model.AuditFilter, writer io.Writer) (rows int, err error)
- func (service *Service) Page(ctx context.Context, filter model.AuditFilter, page int, size int) (model.AuditPage, error)
- func (service *Service) Record(ctx context.Context, log model.AuditLog) (model.AuditLog, error)
- type Sink
- type SinkError
Constants ¶
const ( ActionDiscover = "discover" ActionDiscoverApply = "discover.apply" ActionAuditExport = "audit.export" ActionB5Reconcile = "b5.reconcile" ActionB5Discard = "b5.quarantine.confirm_discard" )
Variables ¶
var ( // ErrInvalidDecision indicates an audit decision outside the frozen set. ErrInvalidDecision = errors.New("invalid audit decision") // ErrExportLimit indicates an export containing more than 100000 rows. ErrExportLimit = errors.New("audit export exceeds 100000 rows") )
Functions ¶
func NewBatchError ¶
func NewBatchError(kind BatchFailureKind, retryable bool, cause error) error
NewBatchError constructs an explicit backend outcome for the coordinator.
Types ¶
type BatchBackend ¶
type BatchBackend interface {
InsertBatch(ctx context.Context, batch []model.AuditLog) ([]model.AuditLog, error)
}
BatchBackend durably commits one batch atomically. Returned rows may be in any order, but each row must carry the EventUUID from its input row.
type BatchError ¶
type BatchError struct {
Kind BatchFailureKind
Retryable bool
Cause error
}
BatchError is the explicit error contract between an S5b adapter and this DB-independent coordinator. Retryable is honored only for transient and global failures; deterministic failures are isolated by stable bisection.
func (*BatchError) Error ¶
func (err *BatchError) Error() string
func (*BatchError) Unwrap ¶
func (err *BatchError) Unwrap() error
type BatchFailureKind ¶
type BatchFailureKind string
BatchFailureKind tells the coordinator what is safe after a failed backend call. Backends must only report rolled-back when commit is known not to have happened.
const ( BatchFailureTransient BatchFailureKind = "transient_rolled_back" BatchFailureDeterministic BatchFailureKind = "deterministic_rolled_back" BatchFailureGlobal BatchFailureKind = "global" BatchFailureCommitUnknown BatchFailureKind = "commit_unknown" )
type GroupCommitSink ¶
type GroupCommitSink struct {
// contains filtered or unexported fields
}
GroupCommitSink multiplexes synchronous Insert calls onto one flusher per physical audit chain.
func NewGroupCommitSink ¶
func NewGroupCommitSink( backend BatchBackend, resolveChainID func(model.AuditLog) string, optionValues ...Option, ) *GroupCommitSink
NewGroupCommitSink constructs an already-started, lazily sharded sink.
func (*GroupCommitSink) Close ¶
func (sink *GroupCommitSink) Close() error
Close stops admission, immediately flushes queued work, and waits for every chain flusher to leave. It never closes the backend itself.
func (*GroupCommitSink) Start ¶
func (sink *GroupCommitSink) Start() error
Start exists for lifecycle symmetry. Construction starts the sink, while this method reports configuration errors before the first Insert.
func (*GroupCommitSink) Stats ¶
func (sink *GroupCommitSink) Stats() GroupCommitStats
Stats returns an aggregate point-in-time snapshot across all physical chains owned by this sink.
type GroupCommitStats ¶
type GroupCommitStats struct {
Batches uint64
BatchEvents uint64
LastBatchSize int64
QueueDepth int
QueueBytes int64
OldestQueueAge time.Duration
TransactionTime time.Duration
Retries uint64
Acknowledgements uint64
AcknowledgementTime time.Duration
Quarantined uint64
ClosingNotStarted uint64
ClosingRolledBack uint64
ClosingCommitUnknown uint64
AdmissionTokens int
Waiters int
CapacityCount int
CapacityBytes int64
ActiveFlusherGoroutines int64
}
GroupCommitStats is an in-process snapshot. S5b can translate it to Prometheus without coupling this coordinator to a metrics implementation.
type Option ¶
type Option func(*groupCommitOptions)
Option configures GroupCommitSink. Options are intentionally small and independently injectable so concurrency tests need not wait on production limits.
func SQLiteGroupCommitOptions ¶
func SQLiteGroupCommitOptions() []Option
SQLiteGroupCommitOptions returns the smaller S5 SQLite queue limits.
func WithAttemptTimeout ¶
func WithCanonicalDigest ¶
func WithDrainTimeout ¶
func WithEnqueueWaitTimeout ¶
func WithFlusherExitedHook ¶
WithFlusherExitedHook installs the ordering point after a chain flusher has completely stopped. An owner may close its backend only after Close returns (and therefore after this hook has run for every chain).
func WithLinger ¶
func WithMaxBatch ¶
func WithMaxEventBytes ¶
func WithMaxWaiters ¶
func WithQueueCapacity ¶
func WithQueueMaxBytes ¶
func WithRootTimeout ¶
type Reader ¶
type Reader interface {
FilteredPage(
ctx context.Context,
filter model.AuditFilter,
page int,
size int,
) (model.AuditPage, error)
}
Reader returns immutable filtered audit pages.
type Recorder ¶
Recorder synchronously records one validated audit event.
func NewRecorder ¶
NewRecorder returns a synchronous recorder backed by sink.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service combines synchronous recording, filtered reads, and export.
func NewService ¶
NewService returns an audit service backed by repository-shaped interfaces.
func (*Service) ExportJSONL ¶
func (service *Service) ExportJSONL( ctx context.Context, filter model.AuditFilter, writer io.Writer, ) (rows int, err error)
ExportJSONL streams filtered audit events as one JSON object per line.
type SinkError ¶
type SinkError struct {
Code model.DBErrorCode `json:"code"`
Reason string `json:"reason"`
RetryAfterMS int `json:"retry_after_ms,omitempty"`
Retryable bool `json:"retryable"`
Decision string `json:"decision"`
}
SinkError is the stable, transport-independent failure returned by the group-commit admission and flushing paths. HTTP mapping belongs to S5b.
Source Files