dbproxy

package
v0.710.6 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package dbproxy provides a db.Querier implementation that transparently routes write operations while keeping reads fast.

Write strategy (secondary instances):

  1. Attempt the write directly against the local RW SQLite connection (WAL mode allows concurrent writes when there is no lock contention).
  2. If SQLite returns BUSY or LOCKED (another writer holds the lock) the operation is forwarded to the primary instance via ZMQ JSON-RPC, which serialises writes as the single authoritative writer.
  3. Only if the proxy call also fails is an error returned to the caller.

Primary instances use the embedded Querier directly; no proxying occurs.

Index

Constants

View Source 
const DefaultHandoverWait = 20 * time.Second

DefaultHandoverWait bounds how long a forwarded write keeps waiting for a primary that is unreachable or handing its role over (drain → lock release → instance.shutdown → a secondary promotes and rebinds the ports). A graceful handover completes in well under a second; the bound also covers most of the 15 s heartbeat gap after a primary is SIGKILLed. One call already in flight when the bound expires may add up to its own timeout.

View Source 
const MethodDBWrite = "db.write"

MethodDBWrite is the JSON-RPC method name for proxied write operations.

View Source 
const MethodExecStatements = "ExecStatements"

MethodExecStatements is the db.write method carrying registered statements.

Variables

View Source 
var DefaultWriteTimeouts = WriteTimeout{
	Default: 5 * time.Second,
	Long:    30 * time.Second,
}

DefaultWriteTimeouts is used when no explicit timeout is provided.

View Source 
var ErrNotRemote = errors.New("dbproxy: not forwarding writes: this instance is the local writer")

ErrNotRemote is returned by the forwarding helpers (WriteWithRetry, ProxyWriteWithResult) when the proxy has no IPC client, i.e. this instance is the primary's local writer. A caller that checked IsRemote() just before the call can still see it when a failover promotion raced the call; it should then perform the write directly, as a primary would.

Functions

func BindPool added in v0.710.6 

func BindPool(db *sql.DB, proxy *DBProxy)

BindPool records that writes on db belong to proxy's IPC topology, for components that only receive the *sql.DB (e.g. the AG-UI adapter, built by entrypoints that pass the shared pool). app.New binds the pool it is given to its DBProxy. A nil proxy removes the binding.

func DispatchWrite added in v0.306.0 

func DispatchWrite(ctx context.Context, q db.Querier, req WriteRequest) (json.RawMessage, error)

DispatchWrite is the exported entry point for dispatching a write request. The writecoordinator calls this from its serialisation loop.

func ForwardWithResult added in v0.710.6 

func ForwardWithResult[R any](ctx context.Context, p *DBProxy, method string, params any) (result R, forwarded bool, err error)

ForwardWithResult is Forward for a write that returns a typed result.

func IsBusyOrLockedError added in v0.710.6 

func IsBusyOrLockedError(err error) bool

IsBusyOrLockedError reports whether err represents a transient SQLite BUSY/LOCKED condition, whether observed directly — e.g. a *sqlite3.Error from a local BeginTx that never passes through mapToWriteError — or after round-tripping through the IPC write channel, where it surfaces as a *WriteError with ErrCodeBusy. Callers that want to retry an idempotent write (e.g. the session indexer's ReplaceSessionEvents) should use this instead of re-deriving the detection logic themselves.

func IsMethodNotSupportedError added in v0.710.6 

func IsMethodNotSupportedError(err error) bool

IsMethodNotSupportedError reports whether err indicates that the primary does not recognise a proxied write method — a version-skew signal (see the mapToWriteError case above for the two shapes this can arrive in). Callers on a secondary should treat this as "the primary predates this write path" and fall back to an older, more broadly supported one rather than treating it as a permanent failure of the specific write attempted. Unlike IsBusyOrLockedError, this only ever matches after the error has round- tripped through mapToWriteError into a *WriteError — a primary (no proxy configured) never calls a write method by name over IPC, so this check is meaningless, and never true, for a primary's own direct calls.

func ProxyWriteWithResult added in v0.326.0 

func ProxyWriteWithResult[R any](ctx context.Context, p *DBProxy, method string, params any) (R, error)

ProxyWriteWithResult forwards a write method and decodes a typed result. It waits out a primary handover (see forwardWithHandoverRetry) but never re-sends a write whose outcome is unknown: typed writes are mostly creates, and a re-sent create that had in fact been applied would fail with a conflict after succeeding.

func RegisterHandlers 

func RegisterHandlers(bus BusRegistrar, q db.Querier)

RegisterHandlers registers the db.write JSON-RPC handler on the given bus. Only the primary instance should call this.

func RegisterHandlersWithCoordinator added in v0.306.0 

func RegisterHandlersWithCoordinator(bus BusRegistrar, s WriteSubmitter)

RegisterHandlersWithCoordinator registers the db.write handler using a WriteSubmitter for serialisation. Both RegisterHandlers and this function can coexist; the coordinator path is opt-in and replaces the direct-dispatch path on the primary.

func RegisterRemembrancesDispatcher added in v0.326.0 

func RegisterRemembrancesDispatcher(d RemembrancesDispatcher)

RegisterRemembrancesDispatcher registers the RAG write dispatcher on the primary. Must be called before any secondary instance attempts to forward remembrances writes. Calling it more than once replaces the previous registration.

func RegisterStatementExecutor added in v0.710.6 

func RegisterStatementExecutor(db *sql.DB)

RegisterStatementExecutor sets the pool the primary executes forwarded statements on (its read-write pool). A primary registers it when it wires its bus, a promoted secondary before starting its bus; pass nil to clear. Without it, ExecStatements answers "unknown write method", which makes the forwarding secondary fall back to a bounded direct retry.

Types

type BusRegistrar added in v0.306.0 

type BusRegistrar interface {
	RegisterMethod(method string, handler ipc.HandlerFunc)
}

BusRegistrar is the minimal interface needed to register dbproxy RPC methods.

type DBProxy 

type DBProxy struct {
	db.Querier // local reads and direct write attempts — embedded interface
	// contains filtered or unexported fields
}

DBProxy implements db.Querier. Reads are served from the embedded local querier. Writes first attempt the local RW connection (WAL mode); on SQLITE_BUSY/LOCKED they are forwarded via ZMQ JSON-RPC to the primary.

When client is nil the proxy behaves identically to the embedded querier (useful for the primary instance itself, which never proxies).

The client is held in an atomic pointer because failover promotion turns a live secondary's proxy into a passthrough (Promote) while other goroutines keep writing through it: every write reads the pointer exactly once and acts on that snapshot.

func New 

func New(local db.Querier, client *ipc.Client, rpcAddr string) *DBProxy

New creates a DBProxy backed by local for reads and direct write attempts. Pass a non-nil client and the primary's rpcAddr to enable write proxying. Pass client=nil for primary instances (writes go directly to the local querier).

func NewWithInstanceID added in v0.306.0 

func NewWithInstanceID(local db.Querier, client *ipc.Client, rpcAddr, instanceID string) *DBProxy

NewWithInstanceID is like New but records the caller's instance ID in every WriteMeta so the primary can attribute writes to the originating secondary.

func ProxyForPool added in v0.710.6 

func ProxyForPool(db *sql.DB) *DBProxy

ProxyForPool returns the proxy bound to db with BindPool, or nil.

func (*DBProxy) CreateFile 

func (p *DBProxy) CreateFile(ctx context.Context, arg db.CreateFileParams) (db.File, error)

func (*DBProxy) CreateMessage 

func (p *DBProxy) CreateMessage(ctx context.Context, arg db.CreateMessageParams) (db.Message, error)

func (*DBProxy) CreateProject 

func (p *DBProxy) CreateProject(ctx context.Context, arg db.CreateProjectParams) (db.Project, error)

func (*DBProxy) CreateSession 

func (p *DBProxy) CreateSession(ctx context.Context, arg db.CreateSessionParams) (db.Session, error)

func (*DBProxy) DeactivateLowestSkill 

func (p *DBProxy) DeactivateLowestSkill(ctx context.Context) error

func (*DBProxy) DeleteFile 

func (p *DBProxy) DeleteFile(ctx context.Context, id string) error

func (*DBProxy) DeleteMessage 

func (p *DBProxy) DeleteMessage(ctx context.Context, id string) error

func (*DBProxy) DeleteProject 

func (p *DBProxy) DeleteProject(ctx context.Context, id string) error

func (*DBProxy) DeleteSession 

func (p *DBProxy) DeleteSession(ctx context.Context, id string) error

func (*DBProxy) DeleteSessionFiles 

func (p *DBProxy) DeleteSessionFiles(ctx context.Context, sessionID string) error

func (*DBProxy) DeleteSessionMessages 

func (p *DBProxy) DeleteSessionMessages(ctx context.Context, sessionID string) error

func (*DBProxy) Forward added in v0.710.6 

func (p *DBProxy) Forward(ctx context.Context, method string, params any, timeout time.Duration) (forwarded bool, err error)

Forward sends a void write to the primary when this proxy is remote. It reports forwarded=false — and the caller must perform the write directly against its local connection — when p is nil, has no client, or lost it to a failover promotion that raced this call (ErrNotRemote). When forwarded is true, err is the outcome of the forwarded write.

This is the single check stores use for "write through the primary or locally", instead of testing p != nil: after Promote a store's proxy is still non-nil but must no longer forward.

func (*DBProxy) IncrementSkillUsage 

func (p *DBProxy) IncrementSkillUsage(ctx context.Context, id string) error

func (*DBProxy) InsertPromptTemplate 

func (p *DBProxy) InsertPromptTemplate(ctx context.Context, arg db.InsertPromptTemplateParams) (db.PromptTemplate, error)

func (*DBProxy) InsertSessionScore 

func (p *DBProxy) InsertSessionScore(ctx context.Context, arg db.InsertSessionScoreParams) (db.SessionScore, error)

func (*DBProxy) InsertSkill 

func (p *DBProxy) InsertSkill(ctx context.Context, arg db.InsertSkillParams) (db.SkillLibrary, error)

func (*DBProxy) IsRemote added in v0.710.6 

func (p *DBProxy) IsRemote() bool

IsRemote reports whether writes may be forwarded to another (primary) instance over IPC. It is false for a proxy built without a client and after Promote. Safe to call on a nil *DBProxy (reports false), so stores can test an optional proxy with a single call.

func (*DBProxy) MarkProjectInitialized 

func (p *DBProxy) MarkProjectInitialized(ctx context.Context, id string) error

func (*DBProxy) ProbePrimary added in v0.326.0 

func (p *DBProxy) ProbePrimary(ctx context.Context) error

ProbePrimary sends an instance.ping JSON-RPC call to the primary with a 2-second timeout and returns nil on success. Returns an error if the primary is unreachable, the call times out, or the proxy is not configured. Always returns nil when this instance is the primary (no proxy needed).

func (*DBProxy) Promote added in v0.710.6 

func (p *DBProxy) Promote() *ipc.Client

Promote atomically turns the proxy into a pure passthrough to its local querier: from now on no write is forwarded over IPC. It is called when this instance wins a failover promotion and its local pool has become the primary's writer. It returns the IPC client the proxy used (nil if it had none) so the caller decides when to close it; writes already in flight keep the client snapshot they read and finish (or fail) against it.

func (*DBProxy) RPCAddr added in v0.710.6 

func (p *DBProxy) RPCAddr() string

RPCAddr returns the primary endpoint writes are currently forwarded to.

func (*DBProxy) SetHandoverWait added in v0.710.6 

func (p *DBProxy) SetHandoverWait(d time.Duration)

SetHandoverWait overrides DefaultHandoverWait for this proxy (tests, or a caller that must fail faster). d <= 0 restores the default.

func (*DBProxy) SetPrimaryResolver added in v0.710.6 

func (p *DBProxy) SetPrimaryResolver(r PrimaryResolver)

SetPrimaryResolver installs the function consulted between forwarding retries to re-point the proxy at the current primary after a failover. Pass nil to remove it.

func (*DBProxy) UpdateFile 

func (p *DBProxy) UpdateFile(ctx context.Context, arg db.UpdateFileParams) (db.File, error)

func (*DBProxy) UpdateMessage 

func (p *DBProxy) UpdateMessage(ctx context.Context, arg db.UpdateMessageParams) error

func (*DBProxy) UpdateProjectLastOpened 

func (p *DBProxy) UpdateProjectLastOpened(ctx context.Context, arg db.UpdateProjectLastOpenedParams) error

func (*DBProxy) UpdateProjectName added in v0.710.6 

func (p *DBProxy) UpdateProjectName(ctx context.Context, id, name string) error

UpdateProjectName renames a project: direct first, forwarded on lock contention like every other sqlc write. The local querier must implement ProjectNameUpdater (*db.Queries does).

func (*DBProxy) UpdateProjectStatus 

func (p *DBProxy) UpdateProjectStatus(ctx context.Context, arg db.UpdateProjectStatusParams) error

func (*DBProxy) UpdateSession 

func (p *DBProxy) UpdateSession(ctx context.Context, arg db.UpdateSessionParams) (db.Session, error)

func (*DBProxy) UpdateSessionACPState added in v0.416.4 

func (p *DBProxy) UpdateSessionACPState(ctx context.Context, arg db.UpdateSessionACPStateParams) error

func (*DBProxy) WriteWithRetry added in v0.326.0 

func (p *DBProxy) WriteWithRetry(ctx context.Context, method string, params any, timeout time.Duration) error

WriteWithRetry forwards a void write with the provided timeout and retry logic. It returns an error wrapping ErrNotRemote when the proxy has no client (see ErrNotRemote for how callers should react).

type PrimaryResolver added in v0.710.6 

type PrimaryResolver func() (rpcAddr string, ok bool)

PrimaryResolver reports the RPC endpoint of the instance currently holding the IPC primary role, or ok=false when it cannot tell (no lock info, lock just released). The runtime wires one that reads the IPC lock file.

type ProjectNameUpdater added in v0.710.6 

type ProjectNameUpdater interface {
	UpdateProjectName(ctx context.Context, id, name string) error
}

ProjectNameUpdater is the project rename write, which is hand-written on *db.Queries and therefore not part of the generated db.Querier interface.

type RemembrancesDispatcher added in v0.326.0 

type RemembrancesDispatcher interface {
	DispatchRemembrancesWrite(ctx context.Context, method string, params json.RawMessage) (json.RawMessage, error)
}

RemembrancesDispatcher defines the contract for dispatching RAG (KB, Events, Code) write operations on the primary instance. It decouples the dbproxy infrastructure package from the rag business-logic packages, avoiding circular imports.

The primary instance registers an implementation via RegisterRemembrancesDispatcher. When dispatchWrite receives a method name not handled by db.Querier, it delegates to this dispatcher (if registered) so that KB, Events and Code indexing writes forwarded from secondary instances are correctly persisted.

type SQLWriter added in v0.710.6 

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

SQLWriter executes registered write statements against db, forwarding them to the IPC primary on lock contention when its proxy is remote. With a nil or non-remote proxy (a primary, a promoted secondary, a CLI) it is exactly a direct ExecContext / transaction on db. Without an explicit proxy it uses the proxy db was bound to with BindPool, if any.

func NewSQLWriter added in v0.710.6 

func NewSQLWriter(db *sql.DB, proxy *DBProxy) *SQLWriter

NewSQLWriter builds a writer over db; proxy may be nil.

func (*SQLWriter) Exec added in v0.710.6 

func (w *SQLWriter) Exec(ctx context.Context, stmt Statement, args ...any) (int64, error)

Exec executes one statement and returns the rows it affected.

func (*SQLWriter) ExecBatch added in v0.710.6 

func (w *SQLWriter) ExecBatch(ctx context.Context, calls ...StmtCall) ([]int64, error)

ExecBatch executes calls atomically (one transaction when there are several) and returns the rows affected by each.

func (*SQLWriter) SetProxy added in v0.710.6 

func (w *SQLWriter) SetProxy(p *DBProxy)

SetProxy sets (or clears, with nil) the write proxy. Safe to call while other goroutines write.

type Statement added in v0.710.6 

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

Statement is a write statement registered under a stable name.

func RegisterStatement added in v0.710.6 

func RegisterStatement(name, query string) Statement

RegisterStatement registers query under name and returns the handle writers execute. Call it from package-level var initialisers so every instance of the binary has the same registry. Registering the same name twice with the same SQL is harmless; with different SQL it panics (a programming error).

func (Statement) Name added in v0.710.6 

func (s Statement) Name() string

Name returns the registered name.

func (Statement) Query added in v0.710.6 

func (s Statement) Query() string

Query returns the SQL text.

func (Statement) With added in v0.710.6 

func (s Statement) With(args ...any) StmtCall

With binds arguments to the statement for SQLWriter.ExecBatch.

type StmtCall added in v0.710.6 

type StmtCall struct {
	Stmt Statement
	Args []any
}

StmtCall is one statement execution in a batch.

type UpdateProjectNameParams added in v0.710.6 

type UpdateProjectNameParams struct {
	ID   string `json:"id"`
	Name string `json:"name"`
}

UpdateProjectNameParams is the forwarded payload of UpdateProjectName.

type WriteError added in v0.306.0 

type WriteError struct {
	Code    WriteErrorCode `json:"code"`
	Message string         `json:"message"`
	Method  string         `json:"method"`
}

WriteError is a structured error returned by the write channel.

func ClassifyError added in v0.710.6 

func ClassifyError(method string, err error) *WriteError

ClassifyError maps an error observed on the write channel (an IPC call failure, or a primary-side error that crossed the boundary as text) to the *WriteError a forwarding secondary would see. Exposed so packages that produce such errors (e.g. the write coordinator) can pin their texts to the intended codes in tests.

func (*WriteError) Error added in v0.306.0 

func (e *WriteError) Error() string

func (*WriteError) IsRetryable added in v0.306.0 

func (e *WriteError) IsRetryable() bool

IsRetryable reports whether the error is transient and the operation may succeed on retry.

type WriteErrorCode added in v0.306.0 

type WriteErrorCode string

WriteErrorCode identifies the category of a write-channel failure. 

const (
	ErrCodeTimeout        WriteErrorCode = "TIMEOUT"
	ErrCodeUnreachable    WriteErrorCode = "UNREACHABLE"
	ErrCodeMethodNotFound WriteErrorCode = "METHOD_NOT_FOUND"
	ErrCodeInvalidParams  WriteErrorCode = "INVALID_PARAMS"
	ErrCodeConflict       WriteErrorCode = "CONFLICT"
	// ErrCodeBusy identifies a transient SQLITE_BUSY/SQLITE_LOCKED failure:
	// the write could not acquire the lock in time (e.g. another writer holds
	// it, or a read->write upgrade collided with a concurrent commit — see
	// [[pando/analysis/session_index_locked_residual_risk.md]] section 2).
	// Unlike ErrCodeInternal, this is always safe to retry for an idempotent
	// write such as ReplaceSessionEvents.
	ErrCodeBusy WriteErrorCode = "BUSY"
	// ErrCodeUnavailable identifies a primary that refused the write without
	// applying it because it is handing its role over (its write coordinator
	// is draining or already shut down). The write is safe to re-send, and
	// the forwarding retry loop waits for the next primary on this code (see
	// forwardWithHandoverRetry).
	ErrCodeUnavailable WriteErrorCode = "UNAVAILABLE"
	ErrCodeInternal    WriteErrorCode = "INTERNAL"
)

type WriteMeta added in v0.306.0 

type WriteMeta struct {
	SourceInstanceID string `json:"source_instance_id"`
	RequestID        string `json:"request_id"`
	Timestamp        string `json:"timestamp"` // RFC3339
}

WriteMeta carries tracing metadata attached to every proxied write request. The primary logs this on each write, making write provenance easy to trace.

type WriteRequest 

type WriteRequest struct {
	Meta   WriteMeta       `json:"meta"`
	Method string          `json:"method"`
	Params json.RawMessage `json:"params"`
}

WriteRequest is the JSON-RPC params struct for a proxied write.

type WriteSubmitter added in v0.306.0 

type WriteSubmitter interface {
	Submit(ctx context.Context, req WriteRequest) (json.RawMessage, error)
}

WriteSubmitter serialises and executes a write request, returning the JSON result. Implemented by writecoordinator.Coordinator to avoid circular imports.

type WriteTimeout added in v0.306.0 

type WriteTimeout struct {
	Default time.Duration
	Long    time.Duration
}

WriteTimeout groups deadline durations for different write categories.

Source Files

View all Source files

Directories

Path Synopsis
Package proxytest builds a real two-role IPC write topology for tests: a migrated database file, the primary's pool with a write coordinator served over a real ZMQ bus, and a secondary's 1-connection 200 ms pool with a DBProxy forwarding to that bus.
 Package proxytest builds a real two-role IPC write topology for tests: a migrated database file, the primary's pool with a write coordinator served over a real ZMQ bus, and a secondary's 1-connection 200 ms pool with a DBProxy forwarding to that bus.

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.