runtime

package
v0.0.0-...-2ac760b Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: MIT Imports: 29 Imported by: 0

README

Runtime

Purpose

runtime owns the shared process wiring used by every Eshu binary at startup. It provides: admin HTTP muxes, health and readiness probes, status metrics endpoints, data-store configuration and connection helpers, retry policy defaults, memory limit tuning, API key resolution, and recovery admin routes. No binary implements this wiring on its own; each calls the helpers here.

Where this fits in the pipeline

flowchart TB
  subgraph Binaries
    A["cmd/api"]
    B["cmd/ingester"]
    C["cmd/reducer"]
    D["cmd/projector"]
    E["cmd/workflow-coordinator"]
  end

  subgraph runtime["internal/runtime"]
    LC["LoadConfig\nNewLifecycle"]
    DS["OpenPostgres\nOpenNeo4jDriver\nLoadGraphBackend"]
    AS["NewStatusAdminServer\nNewStatusAdminMux\nNewAdminMux"]
    RP["LoadRetryPolicyConfig"]
    AK["ResolveAPIKey"]
    ML["ConfigureMemoryLimit"]
  end

  A & B & C & D & E --> LC
  B & C & D --> DS
  B & C & D & E --> AS
  B & C & D --> RP
  A --> AK
  B & C --> ML

Every binary that hosts long-running work also passes through internal/app which calls NewLifecycle and optionally NewStatusAdminServer on behalf of the binary's main function.

Internal flow

The call sequence for a typical long-running binary (cmd/ingester, cmd/reducer, cmd/projector):

flowchart TB
  A["main: telemetry.NewBootstrap"] --> B["LoadConfig(serviceName)"]
  B --> C["ConfigureMemoryLimit(logger)"]
  C --> D["OpenPostgres(ctx, os.Getenv)"]
  D --> E["OpenNeo4jDriver(ctx, os.Getenv)\nor LoadGraphBackend + nornicdb path"]
  E --> F["LoadRetryPolicyConfig(os.Getenv, stage)"]
  F --> G["app.NewHostedWithStatusServer\n  -> NewLifecycle(cfg)\n  -> NewStatusAdminServer(cfg, reader, opts...)"]
  G --> H["Application.Run(ctx)\n  -> Lifecycle.Start\n  -> Runner.Run blocks\n  -> Lifecycle.Stop on exit"]

NewStatusAdminServer delegates to NewStatusAdminMux, which calls NewAdminMux to mount /healthz, /readyz, /admin/status, and /metrics. When WithRecoveryHandler is passed, RecoveryHandler.Mount adds /admin/replay, /admin/refinalize, and /admin/replay-collector-generations to the same mux.

Lifecycle / workflow

Lifecycle (from lifecycle.go:20) holds ServiceName and a telemetry.Bootstrap. Its Start method validates the bootstrap contract; its Run method blocks until the context is canceled via ContextRunner. HTTPServer (from http_server.go:23) also satisfies the Lifecycle interface defined in internal/app — Start opens the TCP listener and serves in the background; Stop gracefully drains with a configurable ShutdownTimeout (default 5 s).

ComposeLifecycles in internal/app chains multiple Lifecycle values (including HTTPServer instances) into one ordered start/stop chain.

Exported surface

Config and env helpers
  • Config — ServiceName, Command, ListenAddr, MetricsAddr; built by LoadConfig(serviceName) which reads ESHU_LISTEN_ADDR (default 0.0.0.0:8080) and ESHU_METRICS_ADDR (default 0.0.0.0:9464)
  • LoadConfig(serviceName) — validates and returns a Config; fails if any field is blank
Data-store helpers
  • GraphBackend — string type; constants GraphBackendNeo4j ("neo4j") and GraphBackendNornicDB ("nornicdb"); LoadGraphBackend reads ESHU_GRAPH_BACKEND, empty defaults to nornicdb, invalid values fail at startup
  • PostgresConfig / PostgresPoolSetter — config struct and interface for pool tuning; loaded by LoadPostgresConfig from ESHU_FACT_STORE_DSN, ESHU_CONTENT_STORE_DSN, or ESHU_POSTGRES_DSN plus optional pool knobs
  • Neo4jConfig — driver and pool tuning; loaded by LoadNeo4jConfig from ESHU_NEO4J_URI / NEO4J_URI, ESHU_NEO4J_USERNAME / NEO4J_USERNAME, ESHU_NEO4J_PASSWORD / NEO4J_PASSWORD, and optional pool knobs
  • OpenPostgres(ctx, getenv) — opens, tunes via ConfigurePostgresPool, and pings a Postgres connection; returns *sql.DB
  • OpenNeo4jDriver(ctx, getenv) — opens a Neo4j/NornicDB Bolt driver, applies ApplyNeo4jConfig, verifies connectivity; returns neo4jdriver.DriverWithContext
  • ConfigurePostgresPool(target, cfg) — applies PostgresConfig to any PostgresPoolSetter
  • ApplyNeo4jConfig(target, cfg) — applies Neo4jConfig to a *neo4jconfig.Config
Admin and HTTP surfaces
  • AdminMuxConfig / NewAdminMux — builds /healthz, /readyz, /admin/status, /metrics routes; optionally mounts a RecoveryHandler; service name required
  • HTTPServer / HTTPServerConfig / NewHTTPServer — one HTTP server with Start/Stop lifecycle; Addr() returns the bound address after Start
  • NewStatusAdminServer(cfg, reader, opts...) — admin HTTPServer backed by the status reader; used by all long-running binaries
  • NewStatusMetricsServer(cfg, reader, opts...) — optional dedicated metrics HTTPServer when MetricsAddr differs from ListenAddr; returns nil when MetricsAddr is empty
  • NewPprofServer(getenv) — opt-in net/http/pprof HTTPServer gated by PprofAddrEnvVar (ESHU_PPROF_ADDR); returns (nil, nil) when unset; port-only inputs (:6060) are rewritten to 127.0.0.1:6060 so the default cannot reach beyond the local host
  • NewStatusAdminMux — lower-level mux builder; combines status handler, metrics handler, optional recovery routes, and optional app handler
  • NewStatusMetricsHandler(serviceName, reader) — Prometheus-style text handler
  • NewCompositeMetricsHandler(statusHandler, prometheusHandler) — merges hand-rolled runtime gauges and OTEL Prometheus output at /metrics
  • StatusAdminOption — option type; constructors: WithRecoveryHandler, WithPrometheusHandler, WithReadinessProbes
  • ReadinessProbe / ReadinessProbesForDependencies(db, driver) / PostgresReadinessProbe(db, timeout) / GraphReadinessProbe(driver, timeout) — dependency-aware /readyz checks. ReadinessProbesForDependencies returns only the probes for wired (non-nil) dependencies; each probe runs concurrently under a bounded timeout and contributes its cause to the /readyz failure body. The status-snapshot probe (Postgres + schema) always runs as the baseline, so default callers keep their existing readiness contract
Recovery admin
  • RecoveryHandler / NewRecoveryHandler(handler) — mounts /admin/replay (POST), /admin/refinalize (POST), and /admin/replay-collector-generations (POST) on the admin mux; delegates to recovery.Handler; replaces the Python write-plane admin surface
Lifecycle and observability
  • Lifecycle / NewLifecycle(cfg) — validates Config, initializes telemetry.Bootstrap, provides Start / Run / Stop
  • ContextRunner — zero-value struct; blocks until context is canceled; used when a binary has no long-running body of its own
  • Observability / NewObservability() — snapshots telemetry.MetricDimensionKeys, telemetry.SpanNames, telemetry.LogKeys at construction time
Retry policy
  • RetryPolicyConfig — MaxAttempts and RetryDelay
  • LoadRetryPolicyConfig(getenv, stagePrefix) — reads ESHU_{STAGE}_MAX_ATTEMPTS (default 3) and ESHU_{STAGE}_RETRY_DELAY (default 30s); both must be positive; stage prefix is required
Memory limits
  • ConfigureMemoryLimit(logger) — sets GOMEMLIMIT from cgroup memory × DefaultMemLimitRatio (0.70), floor MinMemLimit (512 MiB); unconditionally sets GODEBUG=madvdontneed=1; respects explicit GOMEMLIMIT env var as highest priority
API key
  • ResolveAPIKey(getenv) — resolution order: explicit ESHU_API_KEY env, then persisted ESHU_HOME/.env, then auto-generated 32-byte hex token when ESHU_AUTO_GENERATE_API_KEY is truthy; writes generated tokens back to the env file under .env.lock so follow-on CLI and service processes reuse the same token
Status requests
  • StatusRequestStore — interface for durable scan/reindex lifecycle ops
  • StatusRequestHandler / NewStatusRequestHandler(store) — manages RequestScan, ClaimScan, CompleteScan, RequestReindex, ClaimReindex, CompleteReindex
  • RequestState — idle, pending, running, completed, failed
  • ScanRequest / ReindexRequest — lifecycle state structs

Dependencies

Package Used for
internal/buildinfo AppVersion() in runtime metrics labels
internal/recovery recovery.Handler backing RecoveryHandler
internal/status statuspkg.Reader for admin and metrics handlers
internal/telemetry Bootstrap, MetricDimensionKeys, SpanNames, LogKeys, SkippedRefreshCount, DefaultServiceNamespace

Telemetry

This package emits no OTEL spans or traces of its own. The metrics endpoint at /metrics exposes hand-rolled Prometheus-style gauges derived from the statuspkg.Reader. Metric names (all eshu_runtime_ prefix):

  • eshu_runtime_info — binary identity labels (service name, namespace, version)
  • eshu_runtime_scope_active, eshu_runtime_scope_changed, eshu_runtime_scope_unchanged
  • eshu_runtime_refresh_skipped_total
  • eshu_runtime_retry_policy_max_attempts, eshu_runtime_retry_policy_retry_delay_seconds
  • eshu_runtime_health_state — labeled state (healthy/progressing/degraded/stalled)
  • eshu_runtime_queue_total, eshu_runtime_queue_outstanding, and queue depth gauges
  • eshu_runtime_provenance_edge_identity_upgrade_applied and eshu_runtime_provenance_edge_identity_upgrade_required — migration 096 compatibility-fence state and active replay-required work
  • eshu_runtime_stage_items — labeled by stage and status
  • eshu_runtime_domain_outstanding and per-domain backlog gauges
  • eshu_runtime_collector_generation_dead_letter, eshu_runtime_collector_generation_replay_requested, eshu_runtime_collector_generation_replay_attempts, and eshu_runtime_collector_generation_dead_letter_oldest_age_seconds
  • eshu_runtime_coordinator_* — coordinator claim and completeness counters

When WithPrometheusHandler is set, NewCompositeMetricsHandler appends OTEL Prometheus output after the hand-rolled gauges at the same /metrics endpoint.

Operational notes

  • /healthz (liveness) returns 200 OK unconditionally when no AdminCheck is wired; it is intentionally dependency-free so a transient Postgres or graph outage never restarts an otherwise healthy process.

  • /readyz (readiness) runs the status-snapshot probe (Postgres + schema) plus any probes registered via WithReadinessProbes. The API and MCP server register PostgresReadinessProbe (bounded PingContext) and GraphReadinessProbe (bounded Bolt VerifyConnectivity, covering both Neo4j and NornicDB). Each probe runs concurrently under its own bounded timeout; a failure returns 503 with a cause body naming every failing dependency (e.g. graph: ...; postgres: ...). A nil graph driver (local lightweight profile) reports ready so readiness is not gated on an unused dependency.

  • Readiness anti-flap is handled at the Kubernetes probe layer (readinessProbe.failureThreshold), not by in-process state, so the endpoint always reports true current dependency state. See Health And Readiness Probes.

    No-Regression Evidence: dependency probes run only on /readyz hits at the Kubernetes probe cadence (default periodSeconds: 15), never on the query or graph-write hot paths. Each probe is a single bounded connection check (PingContext / VerifyConnectivity, default 2s timeout) executed concurrently; TestRunReadinessProbeBoundsSlowDependency confirms a blocked dependency returns in well under one second. Backend: NornicDB (Bolt) and Postgres via the existing shared pools; no new pool, worker, or queue is introduced. Verified by go test ./internal/runtime ./cmd/api ./cmd/mcp-server -count=1.

    Observability Evidence: /readyz now distinguishes alive-but-broken from ready — the 503 body names the failing dependency and its error, so an operator can tell graph-down from Postgres pool-exhaustion from schema-not-applied (status_snapshot) without shelling into the pod. Liveness (/healthz) stays dependency-free. The Helm readinessProbe.failureThreshold (3) debounces transient blips before pulling the pod from Service endpoints.

  • eshu_runtime_queue_oldest_outstanding_age_seconds aging means workers cannot keep up with ingest rate; investigate worker count and graph backend latency before changing pool sizes.

  • eshu_runtime_health_state{state="stalled"} = 1 means the pipeline is not making progress; check structured logs and failure_class before restarting.

  • eshu_runtime_provenance_edge_identity_upgrade_required > 0 after an old reducer reports successful ACKs means migration 096 is deliberately requeuing incompatible terminal transitions. Roll the reducer forward; do not disable the fence triggers.

  • eshu_runtime_collector_generation_dead_letter > 0 means a collector commit failed before normal projector work items existed. Fix the commit failure, then use /admin/replay-collector-generations with a collector kind and bounded limit to request source-level replay.

  • Admin endpoints have no authentication. They must be bound to the admin port (default 0.0.0.0:9464) and not exposed on the public API port.

  • compose_defaults_test.go enforces that docker-compose.yaml sets ESHU_GRAPH_BACKEND=nornicdb for all graph runtime services and that the telemetry overlay is never mixed into a run without an explicit base file.

  • compose_nornicdb_image_test.go enforces that Compose builds the exact orneryd/NornicDB#290 source commit by default, labels the image with its full revision, uses the build pull policy, and does not force amd64 when the operator leaves the Compose platform override unset.

Extension points

  • StatusAdminOption — add new admin mux behavior by defining a new WithPrometheusHandler-style constructor returning a StatusAdminOption; do not mutate AdminMuxConfig directly
  • PostgresPoolSetter — any *sql.DB-like type satisfies the interface; use ConfigurePostgresPool to apply shared defaults without forking the tuning logic
  • AdminMuxConfig.Health and AdminMuxConfig.Ready — supply custom AdminCheck functions to gate the probes on domain-specific invariants

Gotchas / invariants

  • LoadGraphBackend with an unrecognized value fails at startup, not at first use. data_stores.go:90 is the only valid switch for the backend env var; do not add new backend strings without updating this switch and the NornicDB ADR.
  • OpenNeo4jDriver returns an error when ESHU_GRAPH_BACKEND is not neo4j or nornicdb (data_stores.go:290). Both backends use the same Bolt driver path.
  • NewStatusMetricsServer returns (nil, nil) when MetricsAddr is empty. Callers must handle the nil return; MountStatusServer in internal/app checks this.
  • NewPprofServer returns (nil, nil) when ESHU_PPROF_ADDR is unset or whitespace-only, matching the NewStatusMetricsServer precedent. Callers must check the nil return before calling Start. Port-only inputs are rewritten to 127.0.0.1 to keep the default exposure on loopback; explicit hosts (0.0.0.0, named hosts) are preserved.
  • ConfigureMemoryLimit is a no-op when GOMEMLIMIT is already set as an env var; it logs the existing value and returns 0. Do not call it twice.
  • Admin routes are not authenticated by this package. If the admin port is exposed outside a pod, the operator is responsible for network controls.
  • docs/public/deployment/service-runtimes.md
  • docs/public/run-locally/docker-compose.md
  • docs/public/reference/telemetry/index.md
  • docs/public/reference/local-testing.md
  • ADR: docs/public/reference/backend-conformance.md
  • ADR: docs/public/reference/graph-backend-operations.md

Documentation

Overview

Package runtime provides shared process runtime contracts for Eshu services.

The package owns admin HTTP surfaces, metrics endpoints, the opt-in net/http/pprof endpoint, lifecycle wiring, retry policy defaults, API key checks, auto-generated local API key state, and data-store configuration shared by the API, MCP, ingester, reducer, and helper binaries. Recovery routes include work-item replay, refinalize, and collector generation source-level replay requests.

Index

Constants

View Source 
const (
	// DefaultMemLimitRatio is the fraction of container memory to use as
	// GOMEMLIMIT. 70% leaves headroom for non-heap allocations (goroutine
	// stacks, mmap'd files, cgo, kernel page cache).
	DefaultMemLimitRatio = 0.70

	// MinMemLimit is the floor — never set GOMEMLIMIT below this.
	MinMemLimit = 512 << 20 // 512 MiB
)
View Source 
const PprofAddrEnvVar = "ESHU_PPROF_ADDR"

PprofAddrEnvVar is the env var that controls the opt-in pprof endpoint. Operators set it to bind the runtime profiler; leaving it unset disables the endpoint entirely.

Variables

This section is empty.

Functions

func ApplyNeo4jConfig 

func ApplyNeo4jConfig(target *neo4jconfig.Config, cfg Neo4jConfig)

ApplyNeo4jConfig applies the shared Neo4j tuning policy to a driver config.

func ConfigureMemoryLimit 

func ConfigureMemoryLimit(logger *slog.Logger) int64

ConfigureMemoryLimit sets GOMEMLIMIT based on:

  1. GOMEMLIMIT env var (explicit override via Go runtime, highest priority)
  2. Container cgroup memory limit × DefaultMemLimitRatio
  3. No-op if neither is available (let Go defaults apply)

It also unconditionally sets GODEBUG=madvdontneed=1 which forces the Go runtime to release RSS pages to the OS immediately on GC. This prevents the kernel OOM killer from targeting the container based on inflated RSS.

Returns the applied limit in bytes, or 0 if no limit was set.

func ConfigurePostgresPool 

func ConfigurePostgresPool(target PostgresPoolSetter, cfg PostgresConfig)

ConfigurePostgresPool applies the shared Postgres pool policy to a target.

func IsTruthy 

func IsTruthy(value string) bool

IsTruthy reports whether value is one of the accepted truthy spellings for an Eshu boolean environment-variable flag (case-insensitive, surrounding whitespace trimmed). It is the single source of truth for the "is this escape-hatch/opt-in flag on" question across runtime env-var flags such as ESHU_AUTO_GENERATE_API_KEY and ESHU_MCP_ALLOW_UNAUTHENTICATED.

func NewAdminMux 

func NewAdminMux(cfg AdminMuxConfig) (*http.ServeMux, error)

NewAdminMux builds the shared probe and admin route contract for a runtime.

func NewCompositeMetricsHandler 

func NewCompositeMetricsHandler(statusHandler, prometheusHandler http.Handler) http.Handler

NewCompositeMetricsHandler serves OTEL Prometheus output and the hand-rolled runtime gauges from the same /metrics endpoint.

func NewStatusAdminMux 

func NewStatusAdminMux(
	serviceName string,
	reader statuspkg.Reader,
	appHandler http.Handler,
	opts ...StatusAdminOption,
) (*http.ServeMux, error)

NewStatusAdminMux builds the shared status, metrics, recovery, and optional application routes for a long-running Go runtime.

func NewStatusMetricsHandler 

func NewStatusMetricsHandler(serviceName string, reader statuspkg.Reader) (http.Handler, error)

NewStatusMetricsHandler builds a shared Prometheus-style metrics surface from the same status reader used by the runtime admin report.

func OpenPostgres 

func OpenPostgres(ctx context.Context, getenv func(string) string) (*sql.DB, error)

OpenPostgres opens, tunes, and verifies a Postgres connection for a Go service runtime.

func ResolveAPIKey 

func ResolveAPIKey(getenv func(string) string) (string, error)

ResolveAPIKey returns the runtime API token contract for local compose and operator deployments.

Resolution order:

  1. explicit ESHU_API_KEY environment variable
  2. persisted ESHU_HOME/.env entry
  3. auto-generated token when ESHU_AUTO_GENERATE_API_KEY is truthy

When a token is persisted or generated, it is written back to the .env file so the CLI and follow-on runtimes can reuse the same contract.

Types

type AdminCheck 

type AdminCheck func() error

AdminCheck reports whether a runtime probe is healthy.

type AdminMuxConfig 

type AdminMuxConfig struct {
	ServiceName     string
	Health          AdminCheck
	Ready           AdminCheck
	StatusHandler   http.Handler
	MetricsHandler  http.Handler
	RecoveryHandler *RecoveryHandler
}

AdminMuxConfig defines the shared admin and probe routes for a long-running Go runtime.

type Config 

type Config struct {
	ServiceName string
	Command     string
	ListenAddr  string
	MetricsAddr string
}

Config captures the minimal shared process settings for the Go data-plane bootstrap lane.

func LoadConfig 

func LoadConfig(serviceName string) (Config, error)

LoadConfig builds a validated runtime config for the named service.

func (Config) Validate 

func (c Config) Validate() error

Validate checks the config for the small set of invariants required by the bootstrap lane.

type ContextRunner 

type ContextRunner struct{}

ContextRunner blocks until the parent process context is canceled.

func (ContextRunner) Run 

func (ContextRunner) Run(ctx context.Context) error

Run blocks until the process context is canceled.

type GraphBackend 

type GraphBackend string

GraphBackend names the graph database adapter selected for a Eshu runtime. 

const (
	// GraphBackendNeo4j selects the official Neo4j graph adapter.
	GraphBackendNeo4j GraphBackend = "neo4j"
	// GraphBackendNornicDB selects the official NornicDB graph adapter.
	GraphBackendNornicDB GraphBackend = "nornicdb"
)

func LoadGraphBackend 

func LoadGraphBackend(getenv func(string) string) (GraphBackend, error)

LoadGraphBackend validates the selected graph backend for the current process. Empty uses the NornicDB default.

type HTTPServer 

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

HTTPServer owns one runtime-mounted HTTP server and graceful shutdown path.

func NewHTTPServer 

func NewHTTPServer(cfg HTTPServerConfig) (*HTTPServer, error)

NewHTTPServer validates and freezes the shared HTTP server lifecycle config.

func NewPprofServer 

func NewPprofServer(getenv func(string) string) (*HTTPServer, error)

NewPprofServer builds the opt-in pprof HTTP server for a runtime binary.

When ESHU_PPROF_ADDR is unset or whitespace-only, the function returns (nil, nil); every caller must check for a nil *HTTPServer before calling Start, matching the precedent set by NewStatusMetricsServer.

When the env value supplies only a port (":6060"), the bind host is forced to 127.0.0.1 so a typo or a habit picked up from public listeners does not silently expose profiling endpoints on a routable interface. Explicit hosts — including 0.0.0.0 — are preserved.

func NewStatusAdminServer 

func NewStatusAdminServer(cfg Config, reader statuspkg.Reader, opts ...StatusAdminOption) (*HTTPServer, error)

NewStatusAdminServer builds the shared admin HTTP server for a long-running runtime using the storage-backed status reader seam.

func NewStatusMetricsServer 

func NewStatusMetricsServer(cfg Config, reader statuspkg.Reader, opts ...StatusAdminOption) (*HTTPServer, error)

NewStatusMetricsServer builds the shared dedicated metrics HTTP server for a long-running runtime when a separate metrics address is configured.

func (*HTTPServer) Addr 

func (s *HTTPServer) Addr() string

Addr returns the bound address after Start.

func (*HTTPServer) Start 

func (s *HTTPServer) Start(context.Context) error

Start opens the shared runtime HTTP listener and begins serving in the background.

func (*HTTPServer) Stop 

func (s *HTTPServer) Stop(ctx context.Context) error

Stop gracefully shuts down the shared runtime HTTP server.

type HTTPServerConfig 

type HTTPServerConfig struct {
	Addr            string
	Handler         http.Handler
	ShutdownTimeout time.Duration
}

HTTPServerConfig configures a shared runtime-owned HTTP server lifecycle.

type Lifecycle 

type Lifecycle struct {
	ServiceName string
	Telemetry   telemetry.Bootstrap
}

Lifecycle is the minimal start-run-stop surface shared by the bootstrap lane.

func NewLifecycle 

func NewLifecycle(cfg Config) (Lifecycle, error)

NewLifecycle builds a lifecycle wrapper for the supplied config.

func (Lifecycle) Run 

func (l Lifecycle) Run(ctx context.Context) error

Run blocks until the process context is canceled.

func (Lifecycle) Start 

func (l Lifecycle) Start(context.Context) error

Start performs startup hooks for the service.

func (Lifecycle) Stop 

func (l Lifecycle) Stop(context.Context) error

Stop performs shutdown hooks for the service.

type Neo4jConfig 

type Neo4jConfig struct {
	URI                          string
	Username                     string
	Password                     string
	DatabaseName                 string
	MaxConnectionPoolSize        int
	MaxConnectionLifetime        time.Duration
	ConnectionAcquisitionTimeout time.Duration
	SocketConnectTimeout         time.Duration
	VerifyTimeout                time.Duration
}

Neo4jConfig captures shared driver and pool tuning for Go services that talk to Neo4j.

func LoadNeo4jConfig 

func LoadNeo4jConfig(getenv func(string) string) (Neo4jConfig, error)

LoadNeo4jConfig reads the shared Neo4j config from env.

func OpenNeo4jDriver 

func OpenNeo4jDriver(
	ctx context.Context,
	getenv func(string) string,
) (neo4jdriver.DriverWithContext, Neo4jConfig, error)

OpenNeo4jDriver opens and verifies a Neo4j driver with shared pool tuning.

type Observability 

type Observability struct {
	MetricDimensions []string
	SpanNames        []string
	LogKeys          []string
}

Observability carries the frozen OTEL contract through bootstrap wiring.

func NewObservability 

func NewObservability() Observability

NewObservability snapshots the shared telemetry contract for a service.

type PostgresConfig 

type PostgresConfig struct {
	DSN             string
	MaxOpenConns    int
	MaxIdleConns    int
	ConnMaxLifetime time.Duration
	ConnMaxIdleTime time.Duration
	PingTimeout     time.Duration
}

PostgresConfig captures the shared database and pool tuning used by Go services that talk to Postgres.

func LoadPostgresConfig 

func LoadPostgresConfig(getenv func(string) string) (PostgresConfig, error)

LoadPostgresConfig reads the shared Postgres config from env.

type PostgresPoolSetter 

type PostgresPoolSetter interface {
	SetMaxOpenConns(int)
	SetMaxIdleConns(int)
	SetConnMaxLifetime(time.Duration)
	SetConnMaxIdleTime(time.Duration)
}

PostgresPoolSetter is the minimal tuning surface required from sql.DB.

type ReadinessProbe 

type ReadinessProbe struct {
	// Name labels the dependency in the /readyz cause body (e.g. "postgres").
	Name string
	// Timeout bounds this probe; non-positive values fall back to
	// defaultDependencyReadinessTimeout.
	Timeout time.Duration
	// Check reports the dependency error, or nil when the dependency is ready.
	Check func(ctx context.Context) error
}

ReadinessProbe is a single named dependency check evaluated by /readyz. Each probe is run with a bounded timeout and contributes its cause to the aggregated readiness failure body when it fails.

func GraphReadinessProbe 

func GraphReadinessProbe(driver neo4jdriver.DriverWithContext, timeout time.Duration) ReadinessProbe

GraphReadinessProbe verifies graph backend (Bolt) connectivity for /readyz. The same Bolt driver fronts both Neo4j and NornicDB, so one probe covers both backends. A nil driver (for example the local lightweight profile that disables the graph) reports ready so readiness is not gated on a dependency the service does not use.

func PostgresReadinessProbe 

func PostgresReadinessProbe(db *sql.DB, timeout time.Duration) ReadinessProbe

PostgresReadinessProbe verifies Postgres connectivity for /readyz using a bounded Ping. A blocked Ping surfaces as a deadline-exceeded cause, which distinguishes pool exhaustion or an unreachable database from a schema fault reported by the status snapshot probe.

func ReadinessProbesForDependencies 

func ReadinessProbesForDependencies(db *sql.DB, driver neo4jdriver.DriverWithContext) []ReadinessProbe

ReadinessProbesForDependencies builds the standard dependency readiness probes for a long-running service, omitting dependencies that are not wired. A nil db yields no Postgres probe; a nil graph driver yields no graph probe, so the local lightweight profile that disables the graph stays ready. In production wiring both handles are non-nil, so both dependencies are probed.

type RecoveryHandler 

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

RecoveryHandler provides HTTP endpoints for write-plane recovery operations. It replaces the Python admin refinalize and replay surfaces with Go-owned queue replay rather than direct graph mutation.

func NewRecoveryHandler 

func NewRecoveryHandler(handler *recovery.Handler) (*RecoveryHandler, error)

NewRecoveryHandler constructs the HTTP recovery handler.

func (*RecoveryHandler) Mount 

func (h *RecoveryHandler) Mount(mux *http.ServeMux)

Mount registers recovery routes on the given mux.

type ReindexRequest 

type ReindexRequest struct {
	Ingester    string
	State       RequestState
	RequestedAt time.Time
	ClaimedAt   time.Time
	CompletedAt time.Time
	Error       string
}

ReindexRequest captures the current state of a reindex lifecycle for one ingester.

type RequestState 

type RequestState string

RequestState represents the lifecycle state of a scan or reindex request. 

const (
	// RequestStateIdle means no scan or reindex request is currently active.
	RequestStateIdle RequestState = "idle"
	// RequestStatePending means the request has been stored but not claimed.
	RequestStatePending RequestState = "pending"
	// RequestStateRunning means a runtime has claimed and started the request.
	RequestStateRunning RequestState = "running"
	// RequestStateCompleted means the claimed request finished successfully.
	RequestStateCompleted RequestState = "completed"
	// RequestStateFailed means the claimed request ended with an error.
	RequestStateFailed RequestState = "failed"
)

func (RequestState) Validate 

func (s RequestState) Validate() error

Validate returns an error if the state is not a known value.

type RetryPolicyConfig 

type RetryPolicyConfig struct {
	MaxAttempts int
	RetryDelay  time.Duration
	// MaxRetryDelay caps the exponential backoff term so a high attempt
	// count cannot grow the delay unboundedly. Zero/unset falls back to
	// defaultRetryMaxDelay (1 hour).
	MaxRetryDelay time.Duration
	// JitterFraction scales the random component added on top of the
	// exponential term, relative to RetryDelay. A value of 0 disables
	// jitter entirely (deterministic legacy behavior); the default 0.1
	// matches the formula in issue #4450: rand(0, baseDelay*0.1).
	JitterFraction float64
}

RetryPolicyConfig captures bounded retry settings for one runtime stage.

RetryDelay is the base delay; the actual per-attempt delay grows exponentially with the durable attempt count (baseDelay*(1<<attempt)), capped at MaxRetryDelay, plus a uniform random jitter term drawn from [0, RetryDelay*JitterFraction). Fixed, jitter-free delays let many work items that fail at the same instant reconverge on the identical visible_at and self-reinforce into a retry storm that starves new work (#4450); the exponential term and jitter both exist to break that synchronization.

func LoadRetryPolicyConfig 

func LoadRetryPolicyConfig(getenv func(string) string, stagePrefix string) (RetryPolicyConfig, error)

LoadRetryPolicyConfig reads a bounded retry policy using the supplied stage prefix, for example PROJECTOR or REDUCER.

type ScanRequest 

type ScanRequest struct {
	Ingester    string
	State       RequestState
	RequestedAt time.Time
	ClaimedAt   time.Time
	CompletedAt time.Time
	Error       string
}

ScanRequest captures the current state of a scan lifecycle for one ingester.

type StatusAdminOption 

type StatusAdminOption func(*statusAdminOptions)

StatusAdminOption configures optional behavior on the status admin server.

func WithPrometheusHandler 

func WithPrometheusHandler(h http.Handler) StatusAdminOption

WithPrometheusHandler attaches an OTEL Prometheus exporter handler that is served alongside the existing status-based metrics on /metrics.

func WithReadinessProbes 

func WithReadinessProbes(probes ...ReadinessProbe) StatusAdminOption

WithReadinessProbes registers additional dependency checks that /readyz must pass before reporting ready. The status-snapshot probe (Postgres + schema) always runs as the baseline; these probes extend it, for example to verify graph backend connectivity. Each probe runs under its own bounded timeout and contributes its cause to the readiness failure body.

func WithRecoveryHandler 

func WithRecoveryHandler(rh *RecoveryHandler) StatusAdminOption

WithRecoveryHandler attaches a recovery handler to the admin mux, mounting /admin/replay and /admin/refinalize routes alongside the standard probes.

type StatusRequestHandler 

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

StatusRequestHandler manages scan/reindex lifecycle transitions.

func NewStatusRequestHandler 

func NewStatusRequestHandler(store StatusRequestStore) (*StatusRequestHandler, error)

NewStatusRequestHandler constructs a handler with the given store.

func (*StatusRequestHandler) ClaimReindex 

func (h *StatusRequestHandler) ClaimReindex(ctx context.Context, ingester string) (ReindexRequest, error)

ClaimReindex claims a pending reindex request for the given ingester.

func (*StatusRequestHandler) ClaimScan 

func (h *StatusRequestHandler) ClaimScan(ctx context.Context, ingester string) (ScanRequest, error)

ClaimScan claims a pending scan request for the given ingester.

func (*StatusRequestHandler) CompleteReindex 

func (h *StatusRequestHandler) CompleteReindex(ctx context.Context, ingester string, reindexErr string) error

CompleteReindex marks a running reindex as completed or failed.

func (*StatusRequestHandler) CompleteScan 

func (h *StatusRequestHandler) CompleteScan(ctx context.Context, ingester string, scanErr string) error

CompleteScan marks a running scan as completed or failed.

func (*StatusRequestHandler) RequestReindex 

func (h *StatusRequestHandler) RequestReindex(ctx context.Context, ingester string) error

RequestReindex initiates a reindex request for the given ingester.

func (*StatusRequestHandler) RequestScan 

func (h *StatusRequestHandler) RequestScan(ctx context.Context, ingester string) error

RequestScan initiates a scan request for the given ingester.

type StatusRequestStore 

type StatusRequestStore interface {
	// RequestScan transitions a scan request from idle to pending.
	RequestScan(ctx context.Context, ingester string, now time.Time) error

	// ClaimScanRequest transitions a pending scan to running.
	ClaimScanRequest(ctx context.Context, ingester string, now time.Time) (ScanRequest, error)

	// CompleteScanRequest transitions a running scan to completed or failed.
	CompleteScanRequest(ctx context.Context, ingester string, now time.Time, scanErr string) error

	// RequestReindex transitions a reindex request from idle to pending.
	RequestReindex(ctx context.Context, ingester string, now time.Time) error

	// ClaimReindexRequest transitions a pending reindex to running.
	ClaimReindexRequest(ctx context.Context, ingester string, now time.Time) (ReindexRequest, error)

	// CompleteReindexRequest transitions a running reindex to completed or failed.
	CompleteReindexRequest(ctx context.Context, ingester string, now time.Time, reindexErr string) error

	// GetScanState returns the current scan request state for one ingester.
	GetScanState(ctx context.Context, ingester string) (ScanRequest, error)

	// GetReindexState returns the current reindex request state for one ingester.
	GetReindexState(ctx context.Context, ingester string) (ReindexRequest, error)
}

StatusRequestStore provides the durable scan/reindex request lifecycle operations ported from the Python status_store_db.

Source Files

View all Source files

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.