fasten

package module
v0.0.0-...-e4b67f9 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

README

fasten — Go

Audit + correlation SDK for Go services. v1.0.0-beta.

Install

go get github.com/nerdapplabs/fasten/go

Quickstart

Verified to run as-is on Go 1.22+:

package main

import (
    "context"
    "database/sql"
    "time"

    fasten "github.com/nerdapplabs/fasten/go"
    _ "modernc.org/sqlite" // pure-Go SQLite driver; CGO_ENABLED=0 friendly
)

func main() {
    fasten.MustRegister("user", map[fasten.Code]fasten.Meta{
        "USER_CREATED": {
            Domain: "user", Category: "account", Action: "create",
            Severity: fasten.SevInfo, Description: "New user account",
            Emitter: "auth-service", RetentionClass: fasten.RetLong,
        },
    })
    db, _ := sql.Open("sqlite", "./fasten-audit.db")
    store, _ := fasten.NewSQLiteStore(db, "fasten_audit")

    if err := fasten.Init(fasten.Config{
        ServiceID:  "auth-service",
        NodeID:     "host-01",
        AuditStore: store,
    }); err != nil {
        panic(err)
    }
    defer fasten.Flush(5 * time.Second) // drain pending audit rows on exit

    ctx := fasten.WithRequestID(context.Background(), fasten.MintID())
    fasten.Emit(ctx, "USER_CREATED",
        fasten.Target("u-42"),
        fasten.Actor("admin", "user"),
        fasten.WithDetail(map[string]any{"email": "alice@example.com"}),
    )
    fasten.LogInfo(ctx, "signup_complete", "user_id", "u-42")
}

The audit row + sys log share the same request_id on stdout. The audit row is also persisted to ./fasten-audit.db.

Worked example — net/http service

A minimal HTTP service with X-Request-ID propagation, an audit row per request: see examples/server.go.

cd go/examples
FASTEN_SERVICE_ID=demo FASTEN_NODE_ID=host-01 go run server.go
# in another shell
curl -X POST http://localhost:8080/users -d '{"email":"alice@example.com"}'

Reading logs back: whole record, or a recent window?

Every reader response carries a per-stream completeness flag that answers exactly this:

  • store — the stream is backed by a durable store. The response is a query over the whole recorded history (paged by limit), not a window.
  • ring — the stream lives in a bounded in-memory ring (default 2000 rows, cleared on restart). The response only reaches as far back as the ring: older rows have been evicted, and there is no signal for whether eviction dropped matching rows. Treat it as "recent window", never "the record".
  • store-degraded — store-backed, but at least one row failed to persist (full disk, closed handle, …). Reads still serve from the store, but durable history has known holes. The flag is sticky: it marks the history, not the current sink state.

The flag is the stream's durability class — it never says whether one specific response lost rows. For /correlate, which caps each stream at limit, compare counts (returned) against totals (matching rows available in the backing source): counts < totals means the response is truncated — raise limit or page the per-stream endpoints.

audit reports store when an audit store is attached (the default in production fasten.Init(...)) and ring when the SDK is running stdout-only without one — a stdout-only audit is honestly not a durable record, and completeness must reflect that. api/sys are ring-only unless you attach a StreamStore via Config.APIStore / Config.SyslogStore; persistence is write-through (one synchronous INSERT per pushed row — WAL + synchronous=NORMAL, so no per-commit fsync, but still a per-row disk write; set the pragma in your DSN so every pooled connection gets it).

P1-15: audit-store failure handling

fasten.Emit() defaults to queue mode — rows go onto a bounded channel, drained by a goroutine with exponential backoff (100 ms → 60 s, ±20 % jitter). Store failures stay off the request path. Set Config.AuditStoreFailureStrategy = "raise" to opt into synchronous semantics with *fasten.AuditStoreError. fasten.GetQueueStats() and fasten.Flush(timeout) complete the public surface.

Tests

docker run --rm -v $PWD/go:/work -w /work -e CGO_ENABLED=0 \
  golang:1.22-alpine go test -count=1 ./...

Docs + design

Full reference: https://fasten.sh/docs/ · Design + cross-language design: README.md.

Documentation

Overview

Package fasten — audit + correlation SDK for Go services.

Zero external dependencies. Three streams: syslog, api-log, audit. One request_id carried across every emission via context.

Usage: 

fasten.Register("my-domain", map[fasten.Code]fasten.Meta{
    "MY_CODE": {Action: "created", Severity: fasten.SevInfo, ...},
})
fasten.Init(fasten.Config{ServiceID: "my-svc", NodeID: "node-1", AuditStore: store})
fasten.Emit(ctx, "MY_CODE", fasten.Target("resource/123"))

See ../README.md for the full design.

Index

Constants

View Source 
const (
	MethodHTTP      = "http"       // HTTP/HTTPS request (REST, GraphQL, gRPC-web, webhook)
	MethodMQTT      = "mqtt"       // MQTT message (IoT telemetry, device command)
	MethodCLI       = "cli"        // CLI command typed by a human
	MethodScheduler = "scheduler"  // Automated cron or task scheduler
	MethodUI        = "ui"         // Web or desktop UI action, human-initiated
	MethodAgentTool = "agent_tool" // AI agent tool call
	MethodSDK       = "sdk"        // Direct SDK call, no transport shim active. Default. (default)
)
View Source 
const RedactReplacement = "***"

Variables

View Source 
var Default = &Engine{}

Default is the package-level Engine used by all free-function API calls.

View Source 
var RedactPatterns = []string{
	"api[_-]?key",
	"password",
	"passwd",
	"token",
	"secret",
	"authorization",
	"bearer",
	"m2m[_-]?key",
	"cert[_-]?private",
	"private[_-]?key",
	"access_key",
	"session_id",
	"cookie",
	"credential",
}

RedactPatterns are the default PII field key patterns (case-insensitive regex on keys).

View Source 
var SentinelKinds = fastenctx.SentinelKinds

SentinelKinds re-exports fastenctx.SentinelKinds — the namespaces for rows written outside a real request context (boot, sched, bg, lib, orphan).

Functions

func APILogger 

func APILogger(skipPaths ...string) func(http.Handler) http.Handler

APILogger pushes each inbound HTTP request into fasten's API ring buffer. Skips paths that match any of skipPaths (e.g. "/_system/health").

func Background 

func Background(ctx context.Context) context.Context

Background returns ctx carrying a bg- sentinel request_id when ctx has none, so work outside a request — a background goroutine, worker, or scheduled tick — stays correlatable; if ctx already carries a request_id it is returned unchanged. §5.1/§8.1: keeps the every-row-correlatable invariant for context-less work instead of leaving orphans. Pass "sched"/"lib" via BackgroundKind for those namespaces.

func BackgroundKind 

func BackgroundKind(ctx context.Context, kind string) context.Context

BackgroundKind is Background with an explicit sentinel namespace.

func Flush 

func Flush(timeout time.Duration) bool

Flush blocks until the Default engine's pending rows drain.

func Go 

func Go(ctx context.Context, fn func(context.Context))

Go runs fn in a new goroutine under a Background context: it inherits ctx's request_id if present, else a fresh bg- sentinel, so the goroutine's sys logs remain correlatable. The mirror of Python's fasten.go.

func Init 

func Init(cfg Config) error

Init configures fasten. Delegates to Default.Init.

func IsSentinel 

func IsSentinel(requestID string) bool

IsSentinel reports whether requestID is a minted sentinel, not a real id. Delegates to fastenctx (cgo-free).

func Load 

func Load(path string) error

Load reads a yaml catalog file into the registry.

Errors raise loudly (return error) — designed for startup, restart-safe. Use Reload() to refresh later without restarting the process.

func LogDebug 

func LogDebug(ctx context.Context, event string, kv ...any)

func LogError 

func LogError(ctx context.Context, event string, kv ...any)

func LogInfo 

func LogInfo(ctx context.Context, event string, kv ...any)

LogInfo / LogWarn / LogError / LogDebug write {shape:"sys"} NDJSON lines. Pairs come as key1, value1, key2, value2, ... (slog-style).

func LogWarn 

func LogWarn(ctx context.Context, event string, kv ...any)

func MintID 

func MintID() string

MintID returns a new 12-character hex request id. Delegates to the zero-dependency fastenctx subpackage so a zero-cgo consumer can mint ids without importing the cgo-bound top-level package.

func MintSentinel 

func MintSentinel(kind, serviceID string) string

MintSentinel mints a namespaced sentinel request_id (e.g. "orphan-svc-ab12cd34ef56"). Panics on an unknown kind (a programming error, never runtime input). Delegates to fastenctx (cgo-free).

func MustLoad 

func MustLoad(path string)

MustLoad is like Load but panics on error. Convenient for init().

func MustRegister 

func MustRegister(domain Domain, codes map[Code]Meta)

MustRegister is like Register but panics on error. Safe to call in init().

func NewReader 

func NewReader() http.Handler

NewReader is a package-level shorthand for Default.NewReader().

func RedactDetail 

func RedactDetail(d map[string]any) map[string]any

RedactDetail is the package-level shim that redacts via the Default engine. New code should reach for Engine.redactDetail through a specific Engine instance (multi-tenant callers keep isolated redact config that way).

func Register 

func Register(domain Domain, codes map[Code]Meta) error

Register adds a batch of codes for a domain.

All validation (UPPER_SNAKE_CASE key shape, Meta.ID fill/mismatch, domain match, duplicate detection, pii_in_detail→RetShort) is delegated to fasten-core (Rust) so the logic is canonical across all SDKs. On success the Go-side cache is populated from the Rust-validated state.

Drop Meta.ID in new code; the map key is the single source of truth: 

fasten.Register("user", map[fasten.Code]fasten.Meta{
    "USER_CREATED": {Domain: "user", Action: "create", Severity: fasten.SevInfo, ...},
})

func Reload 

func Reload() error

Reload re-reads every previously-loaded path and atomically swaps the registry.

Atomic + fault-tolerant: parse + validate fully into a fresh map; any error returns without touching the live registry. On success, swap under a lock — concurrent Emit readers see either old or new, never partial state.

Reload is NOT additive: codes removed from the yaml file become unknown (Emit raises). Programmatic Register() codes (not in any yaml file) survive the swap.

func RequestID 

func RequestID(next http.Handler) http.Handler

RequestID is a stdlib http.Handler middleware that mints or honours X-Request-ID and sets it in the context for the duration of the request.

func RequestIDFromContext 

func RequestIDFromContext(ctx context.Context) string

RequestIDFromContext reads the ambient id ("" if unset).

Delegates to fastenctx — same key as WithRequestID above.

func RequestIDKind 

func RequestIDKind(requestID string) string

RequestIDKind classifies a request_id by its namespace: a sentinel kind, or "request" for a real correlation id. Delegates to fastenctx (cgo-free).

func WithRequestID 

func WithRequestID(ctx context.Context, id string) context.Context

WithRequestID returns ctx with the id as the ambient correlation id.

Delegates to the zero-dependency fastenctx subpackage so the context key is identical to the one fastenctx (and the HTTP RequestID middleware) use.

Types

type APIRow 

type APIRow map[string]any

APIRow is a structured API request log entry.

type ActorKind 

type ActorKind string
const (
	ActorUser     ActorKind = "user"     // Human user (browser, mobile, CLI on behalf of a user)
	ActorService  ActorKind = "service"  // Internal service or daemon (default)
	ActorSchedule ActorKind = "schedule" // Cron job or task scheduler
	ActorAgent    ActorKind = "agent"    // AI agent
)

type Anchor 

type Anchor string
const (
	Who         Anchor = "who"
	What        Anchor = "what"
	When        Anchor = "when"
	Where       Anchor = "where"
	Whom        Anchor = "whom"
	How         Anchor = "how"
	Correlation Anchor = "correlation"
)

type AuditRepository 

type AuditRepository interface {
	Insert(ctx context.Context, row Row) error
	Query(ctx context.Context, f Filter) ([]Row, error)
	ListUnshipped(ctx context.Context, limit int) ([]Row, error)
	MarkShipped(ctx context.Context, ids []string) error
	Purge(ctx context.Context, before time.Time, respectUnshipped bool) (int, error)
}

AuditRepository is the durable store contract.

type AuditStoreError 

type AuditStoreError struct{ Err error }

AuditStoreError wraps an underlying store error for the "raise" strategy. Use errors.As to recover the cause: 

var aerr *AuditStoreError
if errors.As(err, &aerr) { /* aerr.Err = sqlite3 / postgres err */ }

func (*AuditStoreError) Error 

func (e *AuditStoreError) Error() string

func (*AuditStoreError) Unwrap 

func (e *AuditStoreError) Unwrap() error

type Chips 

type Chips struct {
	RequestID string
	Filters   map[string]string
	Q         string
}

Chips is the translated query: the three reader primitives side by side. RequestID is the exclusive correlation pivot (when set, correlate rather than filter). Filters are structured exact-match chips composed with AND. Q is the bounded free-text fallback.

func TranslateQuery 

func TranslateQuery(text string) Chips

TranslateQuery translates query text with the default rule translator.

type Code 

type Code string

type Config 

type Config struct {
	ServiceID  string
	NodeID     string
	TenantID   string
	AuditStore AuditRepository

	// FR1: opt-in durable persistence for the api/sys streams. Nil → that
	// stream stays ring-only (default, backward compatible). Construct with
	// NewStreamStore(db, table) for SQLite or NewPostgresStreamStore(db, table)
	// for Postgres; the caller owns the *sql.DB as with AuditStore.
	APIStore    StreamRepository
	SyslogStore StreamRepository

	// FR3: opt-in free-text search (/logs/search and q=). Off by default — it is
	// a linear scan, so it must be explicitly enabled (also via
	// FASTEN_SEARCH_ENABLED). Enabling it without a SyslogStore still yields
	// "search requires sys persistence" at read time.
	SearchEnabled bool

	// Redaction customization (parity with the Python SDK). ExtraRedactKeys are
	// added to the built-in PII key patterns; RedactReplacement overrides the
	// "***" token. Both also read from FASTEN_REDACT_KEYS (comma-separated) /
	// FASTEN_REDACT_REPLACEMENT when the field is empty. Empty = defaults.
	ExtraRedactKeys   []string
	RedactReplacement string

	// P1-15
	AuditStoreFailureStrategy string        // "queue" (default) | "raise"
	QueueCapacity             int           // default 100
	QueueRetryInitial         time.Duration // default 100 * time.Millisecond
	QueueRetryMax             time.Duration // default 60 * time.Second
	DisableQueueJitter        bool          // zero (default) = jitter ON
	QueueDrainMaxAttempts     int           // default 50; row → DLQ after this many failures

	// FR1 retention (spec §1): background age-based purge on the api/sys
	// stream stores. Zero disables. Also read from FASTEN_RETENTION_API and
	// FASTEN_RETENTION_SYSLOG when the field is zero (values there are
	// duration tokens like "7d" / "24h", parsed by time.ParseDuration with
	// day support). Runs the first purge on Init, then every hour.
	RetentionAPI    time.Duration
	RetentionSyslog time.Duration

	// #58 PersistStreams: explicit allowlist of streams the operator has
	// opted into persisting. When non-nil, Init asserts the set matches the
	// streams with stores attached (bidirectional — a named stream without
	// a store or an attached store without the name both fail loudly).
	// When nil, persistence is derived from store attachment (the earlier
	// behaviour, and still the default). "audit" is never valid here —
	// audit persistence is driven by AuditStore. Also read from
	// FASTEN_PERSIST_STREAMS as a comma-separated list.
	PersistStreams []string
}

Config for Init. AuditStore nil → audit rows are written to stdout only.

AuditStoreFailureStrategy (P1-15) governs how Emit reacts when the store rejects a row:

  • "queue" (default) — Emit pushes rows onto a bounded in-memory queue and returns immediately. A background drainer goroutine writes to the store with exponential backoff. Emit blocks only when QueueCapacity (queued + in-flight retry) is saturated.
  • "raise" — Emit calls Insert synchronously and returns the wrapped error (*AuditStoreError). Useful for tests and adopters who want loud failures during configuration debugging.

Falls back to env var FASTEN_AUDIT_STORE_FAILURE_STRATEGY when the field is empty.

type Domain 

type Domain string

Domain is a plain string — adopters define their own vocabulary.

type EmitOption 

type EmitOption func(*Row)

func Actor 

func Actor(a, kind string) EmitOption

func Target 

func Target(t string) EmitOption

func WithDetail 

func WithDetail(d map[string]any) EmitOption

func WithMethod 

func WithMethod(m string) EmitOption

type Engine 

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

Engine holds all runtime state for one fasten deployment context.

The package-level free functions (Init, Emit, …) delegate to Default. Applications that need multiple isolated fasten configurations in one process — multi-tenant services, test isolation — construct Engine instances directly: 

a := &fasten.Engine{}
a.Init(fasten.Config{ServiceID: "tenant-a", AuditStore: storeA})
a.Emit(ctx, "ORDER_PLACED", fasten.Target("order/123"))

func (*Engine) Emit 

func (e *Engine) Emit(ctx context.Context, code Code, opts ...EmitOption) (Row, error)

Emit produces an audit row for a registered code.

func (*Engine) Flush 

func (e *Engine) Flush(timeout time.Duration) bool

Flush blocks until pending audit rows drain, or timeout elapses.

func (*Engine) GetQueueStats 

func (e *Engine) GetQueueStats() *QueueStats

GetQueueStats returns a snapshot of the drainer state, or nil in raise mode.

func (*Engine) GetTransport 

func (e *Engine) GetTransport() *Transport

GetTransport returns the active Transport (ring buffers + stdout). Nil before Init.

func (*Engine) Init 

func (e *Engine) Init(cfg Config) error

func (*Engine) LogSys 

func (e *Engine) LogSys(ctx context.Context, level, event string, kv []any)

LogSys writes a structured {shape:"sys"} line via this Engine.

func (*Engine) NewReader 

func (e *Engine) NewReader() http.Handler

NewReader returns an http.Handler bound to this Engine's configuration (stores, SearchEnabled, persisted streams), serving: 

GET /sys               — syslog ring / store (?q= search when enabled)
GET /api               — api-log ring / store
GET /audit             — audit store query (?request_id=, ?code=, ?domain=,
                         ?since=, ?until=, ?limit=, ?after=<monotonic_seq>)
GET /audit/doctor      — audit pipeline health
GET /search            — cross-stream free-text search (sys-only in v1)
GET /correlate         — request_id → merged audit/api/sys view
GET /topology          — service_id → event aggregation

SECURITY: these endpoints expose internal state (queue stats, init config, raw audit rows, full log payloads). Mount them behind authentication middleware or restrict them to internal network interfaces before exposing to untrusted callers.

Mount with chi: r.Mount("/api/v1/logs", fasten.NewReader()) Mount with stdlib mux: mux.Handle("/api/v1/logs/", http.StripPrefix("/api/v1/logs", fasten.NewReader()))

Per-reader overrides (the equivalent of Python's router(persist_streams=, search_enabled=, store=, transport=)) are achieved by binding the reader to a separately-configured *Engine rather than passing options to NewReader: 

e2 := &fasten.Engine{}
_ = e2.Init(fasten.Config{ServiceID: "svc", NodeID: "n", SearchEnabled: true,
    AuditStore: replica, APIStore: apiStore}) // read-replica / multi-store
mux.Handle("/logs/", http.StripPrefix("/logs", e2.NewReader()))

so a reader can point at different stores or a different search/persistence policy than the process-wide Default engine.

func (*Engine) ResetForTests 

func (e *Engine) ResetForTests()

ResetForTests resets all runtime state to pre-Init defaults. Only for test fixtures — do not call in production code.

func (*Engine) SearchEnabled 

func (e *Engine) SearchEnabled() bool

SearchEnabled reports whether FR3 free-text search is enabled on this engine.

type Filter 

type Filter struct {
	RequestID    string
	Code         Code
	Domain       Domain
	SourceNodeID string
	TenantID     string
	Actor        string
	Target       string
	Since        time.Time
	Until        time.Time
	Limit        int
	// AfterSeq is the canonical cursor: results are newest-first, so paging
	// forward returns older rows — only rows with monotonic_seq < AfterSeq.
	// Set to next_after (the smallest MonotonicSeq of the previous page) to
	// continue. Insert-stable, unlike Offset.
	AfterSeq int64
	// Offset is the alternative page-number cursor (SQL OFFSET), for UIs that
	// want total/limit/offset. Drifts as rows are inserted; prefer AfterSeq.
	Offset int
}

Filter — query parameters for AuditRepository.Query.

type IngestResult 

type IngestResult struct {
	Inserted        int    `json:"inserted"`
	RejectedFromSeq int64  `json:"rejected_from_seq"`
	Reason          string `json:"reason"`
}

IngestResult reports the outcome of an IngestReplicated call.

IngestReplicated inserts the longest VERIFIED prefix of the batch (from the chain start up to, but not including, the first break) in a single transaction and never returns a chain-break error. Inserted counts the rows committed. When the chain breaks partway, RejectedFromSeq carries the MonotonicSeq of the first broken row (rows at/after it are NOT inserted) and Reason explains where it diverged, so the sender resyncs from that point instead of re-shipping a poison-pilled batch forever. On a fully-verified batch RejectedFromSeq is 0 and Reason is "".

type Meta 

type Meta struct {
	ID             Code
	Domain         Domain
	Category       string
	Action         string
	Severity       Severity
	Description    string
	Emitter        string
	RetentionClass RetentionClass
	HighVolume     bool
	PiiInDetail    bool
	// DetailPassthroughKeys: when PiiInDetail=true, only these keys (if any)
	// survive Emit. Everything else is replaced. Empty = scrub everything.
	DetailPassthroughKeys []string
}

Meta is the per-code metadata registered once at startup.

ID is optional in adopter code — Register fills it from the map key at registration time. Setting ID explicitly is allowed but must match the map key (mismatch is a typo, never a feature).

PiiInDetail=true (P1-5) carries three enforced runtime effects:

  1. RetentionClass is forced to RetShort at register-time. Any other declared retention class triggers a WARNING in the registration log.
  2. The Detail payload is force-redacted on Emit, regardless of key names: by default the whole map becomes {"_redacted":"***", "_pii_in_detail": true}. Adopters who genuinely need fields preserved declare them in DetailPassthroughKeys.
  3. Audit rows carry a PiiInDetail bool so retention sweeps and compliance reports can filter PII rows distinctly.

type PostgresStore 

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

PostgresStore is an AuditRepository backed by a *sql.DB connected to PostgreSQL. The caller is responsible for importing the Postgres driver and opening the DB.

Usage: 

import _ "github.com/lib/pq"
db, _ := sql.Open("postgres", os.Getenv("DATABASE_URL"))
store, _ := fasten.NewPostgresStore(db, "public.fasten_audit")

func NewPostgresStore 

func NewPostgresStore(db *sql.DB, tableName string) (*PostgresStore, error)

NewPostgresStore creates and migrates the audit table, then returns the store. tableName may be a plain identifier or schema-qualified (schema.table).

func (*PostgresStore) CountFiltered 

func (s *PostgresStore) CountFiltered(ctx context.Context, f Filter) (int, error)

CountFiltered returns the total number of rows matching the same filter as Query, ignoring Limit — so a capped read (e.g. /correlate) can report how much matching history it truncated.

func (*PostgresStore) Degraded 

func (s *PostgresStore) Degraded() bool

Degraded reports whether at least one persist failure was swallowed. Drives the "store-degraded" audit completeness flag.

func (*PostgresStore) IngestReplicated 

func (s *PostgresStore) IngestReplicated(ctx context.Context, rows []Row) (IngestResult, error)

IngestReplicated verifies and stores a batch of rows replicated from another origin. The verified prefix is inserted in a single transaction; a chain break inserts only the prefix and reports RejectedFromSeq (no error). See ingestReplicatedTx.

func (*PostgresStore) Insert 

func (s *PostgresStore) Insert(ctx context.Context, row Row) error

Insert is a thin alias for InsertOriginated — the engine emit/drainer path.

func (*PostgresStore) InsertOriginated 

func (s *PostgresStore) InsertOriginated(ctx context.Context, row Row) error

InsertOriginated inserts a row this node ORIGINATED (origin_id == id).

func (*PostgresStore) InsertReplicated 

func (s *PostgresStore) InsertReplicated(ctx context.Context, row Row) error

InsertReplicated inserts a sealed row replicated from another origin (autocommit single-row path).

func (*PostgresStore) ListUnshipped 

func (s *PostgresStore) ListUnshipped(ctx context.Context, limit int) ([]Row, error)

func (*PostgresStore) ListUnshippedOriginated 

func (s *PostgresStore) ListUnshippedOriginated(ctx context.Context, serviceID, sourceNodeID string, limit int) ([]Row, error)

ListUnshippedOriginated returns unshipped rows this node ORIGINATED — scoped to (serviceID, sourceNodeID) AND origin_id = id. Replicated rows ingested from another origin are excluded so they are never re-shipped upstream. See SQLiteStore.ListUnshippedOriginated for the rationale.

func (*PostgresStore) MarkShipped 

func (s *PostgresStore) MarkShipped(ctx context.Context, ids []string) error

func (*PostgresStore) NoteWriteFailure 

func (s *PostgresStore) NoteWriteFailure()

NoteWriteFailure records a swallowed persist failure (durable history has a hole). Callers that surface the Insert error don't mark the store degraded.

func (*PostgresStore) Purge 

func (s *PostgresStore) Purge(ctx context.Context, before time.Time, respectUnshipped bool) (int, error)

func (*PostgresStore) Query 

func (s *PostgresStore) Query(ctx context.Context, f Filter) ([]Row, error)

func (*PostgresStore) Search 

func (s *PostgresStore) Search(ctx context.Context, q, since, until string, limit int) ([]Row, error)

Search runs FR3 free-text search over the audit detail column (§4.1). Case-insensitive ILIKE substring, since= bounded, hard-capped by limit, newest-first, no ranking. Result rows carry request_id for /correlate. %/_/\ in q are escaped so they match literally under ESCAPE E'\\'.

type PostgresStreamStore 

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

PostgresStreamStore is the Postgres-backed StreamRepository — the high-volume sibling of StreamStore. It carries the same schema form "1" columns/indexes and the same query/filter/window/completeness/purge/search semantics, so a stream table is portable across backends and the SQLite and Postgres paths agree (and match the Python SDK). Chosen where SQLite's single-writer lock is the bottleneck.

The caller imports a Postgres driver (e.g. github.com/lib/pq) and owns the *sql.DB, exactly as with the audit PostgresStore. This file uses only database/sql, so it adds no driver dependency to the library build.

func NewPostgresStreamStore 

func NewPostgresStreamStore(db *sql.DB, tableName string) (*PostgresStreamStore, error)

NewPostgresStreamStore validates the table name, ensures the schema/table and indexes exist, and returns a ready store. tableName is required — earlier versions defaulted to "syslog" when empty, so a caller wiring an API stream with a blank name silently landed api rows in the syslog table (PR #59 test-coverage gap). Pass "api_log" / "syslog" (or whatever you choose) explicitly.

func (*PostgresStreamStore) Count 

func (s *PostgresStreamStore) Count() (int, error)

func (*PostgresStreamStore) CountMatching 

func (s *PostgresStreamStore) CountMatching(eq map[string]string, since, until string) (int, error)

func (*PostgresStreamStore) Degraded 

func (s *PostgresStreamStore) Degraded() bool

func (*PostgresStreamStore) Insert 

func (s *PostgresStreamStore) Insert(row map[string]any) error

func (*PostgresStreamStore) NoteWriteFailure 

func (s *PostgresStreamStore) NoteWriteFailure()

func (*PostgresStreamStore) Purge 

func (s *PostgresStreamStore) Purge(before string) (int64, error)

func (*PostgresStreamStore) Query 

func (s *PostgresStreamStore) Query(limit int, eq map[string]string, since, until string) ([]map[string]any, error)

func (*PostgresStreamStore) Search 

func (s *PostgresStreamStore) Search(q, since, until string, limit int) ([]map[string]any, error)

func (*PostgresStreamStore) WriteFailures 

func (s *PostgresStreamStore) WriteFailures() int64

WriteFailures exposes the swallowed-insert count (tests only).

type QueueStats 

type QueueStats struct {
	Depth             int     `json:"depth"`
	Capacity          int     `json:"capacity"`
	HighWater         int     `json:"high_water"`
	DrainedTotal      int     `json:"drained_total"`
	RetryCountActive  int     `json:"retry_count_active"`
	InBackoffSeconds  float64 `json:"in_backoff_seconds"`
	LastError         string  `json:"last_error"`
	DeadLetteredTotal int     `json:"dead_lettered_total"`
	DeadLetterDepth   int     `json:"dead_letter_depth"`
	CapacitySemantics string  `json:"capacity_semantics"`
}

QueueStats is the snapshot returned by GetQueueStats(). Depth is total occupied capacity (queued + in-flight retry) — the value that determines whether the next Emit() blocks.

func GetQueueStats 

func GetQueueStats() *QueueStats

GetQueueStats returns the Default engine's drainer snapshot.

type RetentionClass 

type RetentionClass string
const (
	RetShort  RetentionClass = "short"  // Default 30 days
	RetMedium RetentionClass = "medium" // Default 180 days (default)
	RetLong   RetentionClass = "long"   // Default 1095 days (3 years)
)

type RingBuffer 

type RingBuffer[T any] struct {
	// contains filtered or unexported fields
}

RingBuffer is a thread-safe, fixed-capacity buffer. Oldest entries drop when full. Queries return newest-first.

Implemented as a true circular buffer over a fixed-capacity slice. The earlier `rb.buf = rb.buf[1:]` pop strategy retained the original backing array and only advanced the slice header, so the underlying memory grew by one element per push past capacity until GC eventually reclaimed it — effectively unbounded heap growth on a hot syslog path. Circular index keeps the backing array exactly `cap` entries.

func (*RingBuffer[T]) All 

func (rb *RingBuffer[T]) All() []T

All returns a snapshot newest-first.

func (*RingBuffer[T]) Len 

func (rb *RingBuffer[T]) Len() int

func (*RingBuffer[T]) Push 

func (rb *RingBuffer[T]) Push(item T)

type Row 

type Row struct {
	WireVersion  string    `json:"wire_version"`
	ID           string    `json:"id"`
	OriginID     string    `json:"origin_id"`
	MonotonicSeq int64     `json:"monotonic_seq"`
	Timestamp    time.Time `json:"timestamp"`
	Code         Code      `json:"code"`
	Action       string    `json:"action"`
	Severity     Severity  `json:"severity"`
	ServiceID    string    `json:"service_id"`
	SourceNodeID string    `json:"source_node_id"`
	// TenantID is always emitted (null when absent) per the "always emit the
	// key" convention so readers see a consistent shape across SDKs.
	TenantID  *string        `json:"tenant_id"`
	Actor     string         `json:"actor"`
	ActorKind string         `json:"actor_kind"`
	Target    string         `json:"target"`
	Category  string         `json:"category"`
	Domain    Domain         `json:"domain"`
	Method    string         `json:"method"`
	RequestID string         `json:"request_id"`
	Detail    map[string]any `json:"detail"`
	// P1-5: stamped true when the code declares PiiInDetail=true.
	PiiInDetail bool       `json:"pii_in_detail"`
	ShippedAt   *time.Time `json:"shipped_at,omitempty"`
	// CanonicalFormID names the hashed canonical form that sealed this row. "1"
	// is the current (and only) form. It is INCLUDED in the hashed bytes so the
	// form choice is itself tamper-evident; VerifyChain dispatches on it and
	// rejects unknown ids. See verify.go for the form definitions.
	CanonicalFormID string `json:"canonical_form_id,omitempty"`
	// P1-23: tamper-evidence hash chain. PrevHash is the hex SHA-256 of the
	// preceding row in the (service_id, source_node_id) sequence, or "genesis"
	// for the first row. Hash is SHA-256 of canonical JSON of this row with
	// the "hash" key excluded. Rows written before hash-chain support have
	// empty strings; verify_chain skips them.
	PrevHash string `json:"prev_hash,omitempty"`
	Hash     string `json:"hash,omitempty"`
}

Row is the canonical audit row — lossless conversion to CloudEvent / OTel.

func Emit 

func Emit(ctx context.Context, code Code, opts ...EmitOption) (Row, error)

Emit produces an audit row via the Default engine.

func Seal 

func Seal(prevHash string, row Row) Row

Seal returns a copy of row sealed into the chain: it stamps the current CanonicalFormID, sets PrevHash, and computes the self Hash over the canonical form. This is the ONE canonical way to seal a row — Emit goes through it, so Go Emit, Go VerifyChain and Python seal all agree on a single field set.

func (*Row) UnmarshalJSON 

func (r *Row) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes a Row, preserving the EXACT numeric tokens in Detail.

This is load-bearing for cross-language hash compatibility. The canonical row hash is computed over the JSON rendering of Detail, and Python json.dumps renders a whole-number float as "999.0" while Go's default encoder renders the float64 it gets from a plain unmarshal as "999". A Python-sealed row whose detail carries a whole-number float (e.g. a setpoint value 75.0) would then be rejected by VerifyChain on the Go side — a silent cross-language break.

Decoding Detail with UseNumber keeps each value as a json.Number holding its original token ("999.0", "12.5", "7", "1e+20"), which canonicalJSON re-emits verbatim — so Go reproduces Python's rendering byte-for-byte regardless of how the number was written. The rest of the Row decodes normally.

type RuleTranslator 

type RuleTranslator struct{}

RuleTranslator is the deterministic reference translator — the §6.3 smart-box rules. Prose like "yesterday 2pm" is left to a cloud NL model; this parser does not guess time ranges it cannot ground.

func (RuleTranslator) Translate 

func (RuleTranslator) Translate(text string) Chips

type SQLiteStore 

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

SQLiteStore is an AuditRepository backed by a *sql.DB. The caller is responsible for importing the SQLite driver and opening the DB.

Usage: 

import _ "github.com/mattn/go-sqlite3" // or modernc.org/sqlite
db, _ := sql.Open("sqlite3", "./fasten.db")
store := fasten.NewSQLiteStore(db, "fasten_audit")

func NewSQLiteStore 

func NewSQLiteStore(db *sql.DB, tableName string) (*SQLiteStore, error)

NewSQLiteStore creates and migrates the audit table, then returns the store. tableName must be a plain SQL identifier (^[A-Za-z_][A-Za-z0-9_]*$); any other value is rejected to prevent SQL injection — see store.go regex note above.

func (*SQLiteStore) Count 

func (s *SQLiteStore) Count(ctx context.Context) (int, error)

Count returns the total number of rows in the audit table. Used by the /audit/doctor health endpoint to confirm store reachability.

func (*SQLiteStore) CountFiltered 

func (s *SQLiteStore) CountFiltered(ctx context.Context, f Filter) (int, error)

CountFiltered returns the total number of rows matching the same filter as Query, ignoring Limit — so a capped read (e.g. /correlate) can report how much matching history it truncated.

func (*SQLiteStore) Degraded 

func (s *SQLiteStore) Degraded() bool

Degraded reports whether at least one persist failure was swallowed. Drives the "store-degraded" audit completeness flag.

func (*SQLiteStore) IngestReplicated 

func (s *SQLiteStore) IngestReplicated(ctx context.Context, rows []Row) (IngestResult, error)

IngestReplicated verifies and stores a batch of rows replicated from another origin (node -> upstream aggregator reverse sync). The verified prefix is inserted in a single transaction; a chain break inserts only the prefix and reports RejectedFromSeq (no error). See ingestReplicatedTx.

Decoupled from the Engine: this is store-scoped (a method on the store), so a replication sink calls it with only a store — no fasten.Init, no service identity, no drainer required: 

store, _ := fasten.NewSQLiteStore(db, "audit")
res, _ := store.IngestReplicated(ctx, rows) // no fasten.Init needed

func (*SQLiteStore) Insert 

func (s *SQLiteStore) Insert(ctx context.Context, row Row) error

Insert is a thin alias for InsertOriginated — the engine emit/drainer path. Kept so the AuditRepository interface and the drainer stay stable.

func (*SQLiteStore) InsertOriginated 

func (s *SQLiteStore) InsertOriginated(ctx context.Context, row Row) error

InsertOriginated inserts a row this node ORIGINATED (origin_id == id). Used by the engine's own Emit path.

func (*SQLiteStore) InsertReplicated 

func (s *SQLiteStore) InsertReplicated(ctx context.Context, row Row) error

InsertReplicated inserts a sealed row replicated from another origin. Used by IngestReplicated after the chain verifies (autocommit single-row path).

func (*SQLiteStore) ListUnshipped 

func (s *SQLiteStore) ListUnshipped(ctx context.Context, limit int) ([]Row, error)

func (*SQLiteStore) ListUnshippedOriginated 

func (s *SQLiteStore) ListUnshippedOriginated(ctx context.Context, serviceID, sourceNodeID string, limit int) ([]Row, error)

ListUnshippedOriginated returns unshipped rows this node ORIGINATED — scoped to (serviceID, sourceNodeID) AND origin_id = id. Replicated rows ingested from another origin are excluded: re-shipping them upstream would duplicate another node's sub-chain. This is the originated-only counterpart a relay should use; plain ListUnshipped returns every unshipped row regardless of origin.

func (*SQLiteStore) MarkShipped 

func (s *SQLiteStore) MarkShipped(ctx context.Context, ids []string) error

func (*SQLiteStore) MaxMonotonicSeq 

func (s *SQLiteStore) MaxMonotonicSeq(ctx context.Context, serviceID, sourceNodeID string) (int64, error)

MaxMonotonicSeq returns the maximum monotonic_seq stored for the given (serviceID, sourceNodeID) sub-chain, for seq seeding on restart.

The tamper chain is per (service_id, source_node_id) and monotonic_seq is a per-node counter, so seeding the engine's seq MUST be scoped to the engine's OWN identity. An unscoped MAX() across all origins would, after this node ingested replicated rows from another origin, seed seq from a foreign sub-chain and break this node's own chain. When serviceID and sourceNodeID are both empty the query falls back to the legacy global MAX (used only where identity is unavailable).

func (*SQLiteStore) NoteWriteFailure 

func (s *SQLiteStore) NoteWriteFailure()

NoteWriteFailure records a persist failure the caller swallowed on the hot path (durable history has a hole). Callers that surface the Insert error don't mark the store degraded.

func (*SQLiteStore) Purge 

func (s *SQLiteStore) Purge(ctx context.Context, before time.Time, respectUnshipped bool) (int, error)

func (*SQLiteStore) Query 

func (s *SQLiteStore) Query(ctx context.Context, f Filter) ([]Row, error)

func (*SQLiteStore) Search 

func (s *SQLiteStore) Search(ctx context.Context, q, since, until string, limit int) ([]Row, error)

Search runs FR3 free-text search over the audit store's detail column (§4.1). Case-insensitive substring, since= bounded, hard-capped by limit, newest-first, no relevance ranking. Result rows carry request_id for /correlate follow-up. %/_/\ in q are escaped so they match literally.

func (*SQLiteStore) Sources 

func (s *SQLiteStore) Sources(ctx context.Context, since, until time.Time) ([]map[string]any, error)

Sources aggregates the fleet topology from the rows already recorded: one entry per distinct (source_node_id, service_id, tenant_id) with its row count and first/last-seen timestamps, ordered by count. No separate topology table — the view falls out of the audit rows, so it can't drift. Mirrors the Python SQLiteStore.sources. Optional [since, until] windows the aggregation.

type SeqSeeder 

type SeqSeeder interface {
	MaxMonotonicSeq(ctx context.Context, serviceID, sourceNodeID string) (int64, error)
}

SeqSeeder is an optional extension of AuditRepository. Stores that implement it allow Init() to seed monotonic_seq from persisted rows so post-restart rows never collide on (timestamp, seq) with pre-restart rows.

The seed MUST be scoped to the engine's own (serviceID, sourceNodeID): the tamper chain is per-node and monotonic_seq is a per-node counter, so seeding from an unscoped global MAX would break this node's own chain once it has ingested replicated rows from another origin.

type Severity 

type Severity string

── FASTEN GENERATED ─ source: spec/row-schema.json ─ run: python spec/codegen.py ── 

const (
	SevDebug    Severity = "debug"    // Low-level diagnostic, filtered in production
	SevInfo     Severity = "info"     // Normal operational event (default)
	SevWarn     Severity = "warn"     // Potentially problematic, not yet an error
	SevError    Severity = "error"    // Operation failed, requires attention
	SevCritical Severity = "critical" // Severe failure, may impact availability
)

type SlogHandler 

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

SlogHandler wraps an underlying slog.Handler and pushes each log record into fasten's syslog ring buffer. The underlying handler still writes to its own destination (e.g. stdout JSON) — no double-write occurs.

Usage: 

base := slog.NewJSONHandler(os.Stdout, nil)
logger := slog.New(fasten.NewSlogHandler(base))
slog.SetDefault(logger)

func NewSlogHandler 

func NewSlogHandler(next slog.Handler) *SlogHandler

NewSlogHandler wraps next and pushes to the global fasten transport. If fasten is not yet initialised, the handler degrades to next-only.

func (*SlogHandler) Enabled 

func (h *SlogHandler) Enabled(ctx context.Context, level slog.Level) bool

func (*SlogHandler) Handle 

func (h *SlogHandler) Handle(ctx context.Context, r slog.Record) error

func (*SlogHandler) WithAttrs 

func (h *SlogHandler) WithAttrs(attrs []slog.Attr) slog.Handler

func (*SlogHandler) WithGroup 

func (h *SlogHandler) WithGroup(name string) slog.Handler

type StreamQuery 

type StreamQuery struct {
	Level     string // sys
	ServiceID string // sys
	Event     string // sys
	Method    string // api
	Path      string // api
	Status    string // api ("" = no filter; matched against the row's value)
	RequestID string // common
	Since     string // common — timestamp >= Since (ISO-8601)
	Until     string // common — timestamp <= Until (ISO-8601)
}

StreamQuery holds the optional filters for a sys/api stream read. Empty fields are ignored. The indexed structured fields (event, status, time window) are honoured identically whether the read is served from the ring or the durable store.

The Since/Until window compares timestamps as strings (lexicographic, both in the ring scan and in SQL). That is correct for the canonical UTC form fasten's own writers stamp (fixed-width microseconds + always-Z; see canonical_ts.go), which sorts byte-for-byte identically across Python and Go writers. Adopter-supplied timestamps must use canonical_ts / canonicalTS too, or the window will mis-compare (spec §4.3).

type StreamRepository 

type StreamRepository interface {
	Insert(row map[string]any) error
	Query(limit int, eq map[string]string, since, until string) ([]map[string]any, error)
	CountMatching(eq map[string]string, since, until string) (int, error)
	Count() (int, error)
	Purge(before string) (int64, error)
	Search(q, since, until string, limit int) ([]map[string]any, error)
	NoteWriteFailure()
	Degraded() bool
}

StreamRepository is the durable backing for one ring-buffered stream (api or sys). *StreamStore (SQLite) and *PostgresStreamStore both implement it, so Config/Transport can hold either backend behind one type — mirroring how AuditRepository abstracts the audit store. WriteFailures() is intentionally not in the interface (only tests reach for the concrete count).

type StreamStore 

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

StreamStore is a durable, queryable SQLite backing for one ring-buffered stream (api or sys). Unlike the audit store (typed Row + tamper-evident hash chain), stream rows are schemaless maps produced by the logging/HTTP shims. The full row is persisted as a JSON payload and the queryable fields are duplicated into indexed columns, so the reader can filter by request_id / time / structured fields against durable history instead of a bounded ring.

Table per stream — api and sys never share rows. Rows return newest-first, reconstructed from the stored JSON payload, so a store read is equivalent to a ring read in content and ordering. Note JSON decoding normalises number types: all numbers decode to float64 (a ring read returns the original int), so the value is JSON-equivalent, not type-identical.

Write cost: persistence is write-through — every pushed row is one synchronous INSERT in its own transaction, on the caller's thread. With WAL + synchronous=NORMAL (set as DSN pragmas by OpenStreamStore) a commit is a WAL append without a per-commit fsync, which keeps a hot api/sys stream viable. Even so this is a per-row disk write: for very hot streams prefer ring-only mode. An async drainer may land later as a follow-up if benchmarks show write-through cost is the bottleneck (the audit path already has one).

The caller imports the SQLite driver and opens the *sql.DB, exactly as for NewSQLiteStore. See *PostgresStreamStore for the Postgres backend.

func NewStreamStore 

func NewStreamStore(db *sql.DB, tableName string) (*StreamStore, error)

NewStreamStore creates and migrates a per-stream table on an already-opened *sql.DB, then returns the store. tableName must be a plain SQL identifier (see validIdentifierRe).

Caveat: migrate sets PRAGMA synchronous=NORMAL, but that pragma is per-connection and database/sql pools connections — set it in the DSN so every pooled connection gets it (e.g. mattn/go-sqlite3: "file:app.db?_synchronous=NORMAL"; modernc.org/sqlite: "file:app.db?_pragma=synchronous(NORMAL)"). OpenStreamStore does this automatically for the modernc driver.

func OpenStreamStore 

func OpenStreamStore(dsn, tableName string) (*StreamStore, error)

OpenStreamStore opens a SQLite-backed stream store from a DSN and migrates it, setting WAL + synchronous=NORMAL as DSN pragmas so *every* pooled connection gets them. This is the fix for NewStreamStore's caveat below: a PRAGMA issued during migrate lands on a single pooled connection, and the other connections database/sql opens keep the default synchronous=FULL (an fsync per commit). Setting the pragma in the DSN makes modernc.org/sqlite apply it on every connection open — mirroring the Python StreamStore, which sets the pragma on each connection it opens.

It uses the modernc.org/sqlite driver (pure Go, already the SDK's sqlite dependency). Callers who need a different driver (e.g. mattn/go-sqlite3), or who already own a *sql.DB, should use NewStreamStore(db, tableName) and set the per-connection pragma in their own DSN.

func (*StreamStore) Count 

func (s *StreamStore) Count() (int, error)

Count returns the number of persisted rows (test/diagnostic helper).

func (*StreamStore) CountMatching 

func (s *StreamStore) CountMatching(eq map[string]string, since, until string) (int, error)

CountMatching returns the total number of rows matching the same filters as Query, without a limit — so a capped read can report how much history it truncated (the /correlate totals).

func (*StreamStore) Degraded 

func (s *StreamStore) Degraded() bool

Degraded reports whether durable history has known holes (at least one swallowed persist failure). Drives the "store-degraded" completeness flag.

func (*StreamStore) Insert 

func (s *StreamStore) Insert(row map[string]any) error

Insert write-through persists a single stream row. Newest rows sort first.

func (*StreamStore) NoteWriteFailure 

func (s *StreamStore) NoteWriteFailure()

NoteWriteFailure records that a persist attempt for this store was swallowed. Called by the transport at the swallow point, so callers that handle an Insert error themselves don't mark the store degraded.

func (*StreamStore) Purge 

func (s *StreamStore) Purge(before string) (int64, error)

Purge deletes rows older than before (a canonical UTC ISO-8601 string) and returns how many were removed. Per-stream retention pruning (FR1): api/sys run 10-100x audit volume, so their history is trimmed by age. Unlike audit, stream rows have no ship/replication lifecycle, so this is an unconditional age-based delete backed by idx_<table>_ts.

Rows with a NULL/absent timestamp are never purged (timestamp < ? is never true for NULL) — the age of a timestamp-less row is unknown, so it is kept rather than silently dropped. The comparison is lexicographic, matching the read window; keep before in the same canonical UTC form as stored timestamps.

func (*StreamStore) Query 

func (s *StreamStore) Query(limit int, eq map[string]string, since, until string) ([]map[string]any, error)

Query returns up to limit rows newest-first, filtered by equality on the given indexed columns plus an optional [since, until] window on timestamp. Mirrors the ring's exact-match filter semantics so store and ring agree.

func (*StreamStore) Search 

func (s *StreamStore) Search(q, since, until string, limit int) ([]map[string]any, error)

Search is the FR3 escape hatch: a case-insensitive substring scan over the persisted row text inside a mandatory [since, until] window, newest-first, hard-capped, with NO relevance ranking. For "I only have an error string" when structured field discovery can't help. q is matched against the full serialized row (payload); LIKE wildcards (% _ \) are escaped so they match literally. Mirrors the Python StreamStore.search contract.

func (*StreamStore) WriteFailures 

func (s *StreamStore) WriteFailures() int64

WriteFailures returns how many persist attempts were swallowed.

type SyslogRow 

type SyslogRow map[string]any

SyslogRow is a structured syslog entry.

type Translator 

type Translator interface {
	Translate(text string) Chips
}

Translator turns query text into Chips. The rule-based OSS translator and a cloud NL model are interchangeable behind it.

type Transport 

type Transport struct {
	Syslog      *RingBuffer[SyslogRow]
	API         *RingBuffer[APIRow]
	SyslogStore StreamRepository
	APIStore    StreamRepository
	// contains filtered or unexported fields
}

Transport owns the ring buffers and stdout writes. Syslog and API use push-only (no stdout duplication — caller owns stdout). Audit writes to stdout for belt-and-braces capture.

When a stream store is attached (FR1, opt-in persistence) the ring stays the hot write path and the store is a write-through sink; reads for that stream are served from the store so they reach durable history instead of a bounded window. Without a store the stream is ring-only (default).

func GetTransport 

func GetTransport() *Transport

GetTransport returns the active transport of the Default engine.

func NewTransport 

func NewTransport(maxlen int) *Transport

func (*Transport) APIDepth 

func (t *Transport) APIDepth() int

APIDepth returns the current number of API-log rows in the ring buffer.

func (*Transport) CountAPI 

func (t *Transport) CountAPI(q StreamQuery) (int, error)

CountAPI returns the total number of API rows matching q in the backing source (durable store, or the ring's current window) — the uncapped counterpart of QueryAPI, so capped reads can report truncation.

func (*Transport) CountSyslog 

func (t *Transport) CountSyslog(q StreamQuery) (int, error)

CountSyslog returns the total number of syslog rows matching q in the backing source (durable store, or the ring's current window) — the uncapped counterpart of QuerySyslog, so capped reads can report truncation.

func (*Transport) PushAPI 

func (t *Transport) PushAPI(row APIRow)

PushAPI buffers an API request row (+ persists if a store is attached). No stdout write — middleware owns logging. Persist-failure handling is the same as PushSyslog: swallow, but degrade the completeness flag.

func (*Transport) PushSyslog 

func (t *Transport) PushSyslog(row SyslogRow)

PushSyslog buffers a syslog row (+ persists if a store is attached). No stdout write — caller (slog handler) owns that.

A persist failure must never break the hot logging path (the ring already holds the row), so it is logged to stderr and swallowed — but it is also recorded on the store, which degrades the stream's completeness flag to "store-degraded". Without that, reads (served from the store, never the ring, once a store is attached) would silently miss the row while the flag asserted durability.

func (*Transport) QueryAPI 

func (t *Transport) QueryAPI(limit int, q StreamQuery) ([]APIRow, error)

QueryAPI returns up to limit API rows, newest-first. Served from the durable store when one is attached, otherwise from the in-memory ring.

func (*Transport) QuerySyslog 

func (t *Transport) QuerySyslog(limit int, q StreamQuery) ([]SyslogRow, error)

QuerySyslog returns up to limit syslog rows, newest-first. Served from the durable store when one is attached, otherwise from the in-memory ring.

func (*Transport) SearchAPI 

func (t *Transport) SearchAPI(q, since, until string, limit int) ([]APIRow, bool, error)

SearchAPI runs FR3 free-text search over persisted api history. Same shape as SearchSyslog — the api StreamStore is identical to sys, so the reader plugs both into /search?streams=... without stream-specific branches. Returns (nil, false, nil) when the api stream is ring-only.

func (*Transport) SearchSyslog 

func (t *Transport) SearchSyslog(q, since, until string, limit int) ([]SyslogRow, bool, error)

SearchSyslog runs the FR3 free-text search over persisted sys history. Returns (nil, nil) when the sys stream is ring-only — search is defined over durable history, not a bounded window, so without a store there is nothing honest to search (the reader surfaces this as "search requires sys persistence"). The bool return distinguishes "no store" from "no matches".

func (*Transport) SyslogDepth 

func (t *Transport) SyslogDepth() int

SyslogDepth returns the current number of syslog rows in the ring buffer.

func (*Transport) WriteAudit 

func (t *Transport) WriteAudit(row map[string]any)

WriteAudit writes an audit row to stdout as {"shape":"audit",...} JSON.

type VerifyResult 

type VerifyResult struct {
	OK           bool   `json:"ok"`
	TotalRows    int    `json:"total_rows"`
	FirstBreakAt int64  `json:"first_break_at"`
	Reason       string `json:"reason"`
}

VerifyResult is the outcome of VerifyChain.

OK is true when every hash-bearing row's recomputed hash matches its stored hash AND each row's PrevHash matches the preceding row's Hash. FirstBreakAt is the MonotonicSeq of the first row that fails either check (0 when OK).

func VerifyChain 

func VerifyChain(rows []Row) VerifyResult

VerifyChain walks the per-row tamper-evidence hash chain and reports whether it is intact. It is the Go counterpart of the Python fasten.verify_chain and produces byte-for-byte identical row hashes, so a Go upstream aggregator can verify chains that were sealed by a Python node.

The canonical hashed form, the cross-language rendering rules (including the whole-number-float hazard in §1.3), the canonical_form_id registry, and the replication contract are normative in ../spec/chain-replication.md.

Semantics mirror Python engine.py verify_chain exactly:

  • Rows are sorted by MonotonicSeq before walking.
  • Rows with an empty Hash (written before hash-chain support) are skipped.
  • For each hash-bearing row the canonical row hash is recomputed and compared to the stored Hash.
  • Each row's PrevHash is compared to the preceding row's Hash.
  • On the first mismatch the walk stops and returns OK=false with FirstBreakAt set to that row's MonotonicSeq.

It does NOT detect tail truncation — that needs an external tip anchor.

Source Files

View all Source files

Directories

Path Synopsis
Minimal net/http service wired to fasten.
 Minimal net/http service wired to fasten.
Package fastenctx carries the fasten correlation request-id through a context.Context.
 Package fastenctx carries the fasten correlation request-id through a context.Context.

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.