dbproxy

Documentation

Overview

Package dbproxy provides a db.Querier implementation that transparently routes write operations to the primary Pando instance via ZMQ JSON-RPC, while serving reads from the local (possibly read-only) SQLite database.

Secondary instances use DBProxy so they never write to SQLite directly, preserving the single-writer invariant required by SQLite.

Index

• Constants
• Variables
• func DispatchWrite(ctx context.Context, q db.Querier, req WriteRequest) (json.RawMessage, error)
• func RegisterHandlers(bus BusRegistrar, q db.Querier)
• func RegisterHandlersWithCoordinator(bus BusRegistrar, s WriteSubmitter)
• type BusRegistrar
• type DBProxy
  • func New(local db.Querier, client *ipc.Client, rpcAddr string) *DBProxy
  • func NewWithInstanceID(local db.Querier, client *ipc.Client, rpcAddr, instanceID string) *DBProxy
  • func (p *DBProxy) CreateFile(ctx context.Context, arg db.CreateFileParams) (db.File, error)
  • func (p *DBProxy) CreateMessage(ctx context.Context, arg db.CreateMessageParams) (db.Message, error)
  • func (p *DBProxy) CreateProject(ctx context.Context, arg db.CreateProjectParams) (db.Project, error)
  • func (p *DBProxy) CreateSession(ctx context.Context, arg db.CreateSessionParams) (db.Session, error)
  • func (p *DBProxy) DeactivateLowestSkill(ctx context.Context) error
  • func (p *DBProxy) DeleteFile(ctx context.Context, id string) error
  • func (p *DBProxy) DeleteMessage(ctx context.Context, id string) error
  • func (p *DBProxy) DeleteProject(ctx context.Context, id string) error
  • func (p *DBProxy) DeleteSession(ctx context.Context, id string) error
  • func (p *DBProxy) DeleteSessionFiles(ctx context.Context, sessionID string) error
  • func (p *DBProxy) DeleteSessionMessages(ctx context.Context, sessionID string) error
  • func (p *DBProxy) IncrementSkillUsage(ctx context.Context, id string) error
  • func (p *DBProxy) InsertPromptTemplate(ctx context.Context, arg db.InsertPromptTemplateParams) (db.PromptTemplate, error)
  • func (p *DBProxy) InsertSessionScore(ctx context.Context, arg db.InsertSessionScoreParams) (db.SessionScore, error)
  • func (p *DBProxy) InsertSkill(ctx context.Context, arg db.InsertSkillParams) (db.SkillLibrary, error)
  • func (p *DBProxy) MarkProjectInitialized(ctx context.Context, id string) error
  • func (p *DBProxy) UpdateFile(ctx context.Context, arg db.UpdateFileParams) (db.File, error)
  • func (p *DBProxy) UpdateMessage(ctx context.Context, arg db.UpdateMessageParams) error
  • func (p *DBProxy) UpdateProjectLastOpened(ctx context.Context, arg db.UpdateProjectLastOpenedParams) error
  • func (p *DBProxy) UpdateProjectStatus(ctx context.Context, arg db.UpdateProjectStatusParams) error
  • func (p *DBProxy) UpdateSession(ctx context.Context, arg db.UpdateSessionParams) (db.Session, error)
• type WriteError
  • func (e *WriteError) Error() string
  • func (e *WriteError) IsRetryable() bool
• type WriteErrorCode
• type WriteMeta
• type WriteRequest
• type WriteSubmitter
• type WriteTimeout

Constants

const MethodDBWrite = "db.write"

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

Variables

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

DefaultWriteTimeouts is used when no explicit timeout is provided.

Functions

func DispatchWrite

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 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

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.

Types

type BusRegistrar

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 — embedded interface
	// contains filtered or unexported fields
}

DBProxy implements db.Querier. Reads are served from the embedded local querier. Writes are forwarded via ZMQ JSON-RPC to the primary instance.

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

func New

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

New creates a DBProxy backed by local for reads. 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

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 (*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) 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) MarkProjectInitialized

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

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) 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)

type WriteError

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 (*WriteError) Error

func (e *WriteError) Error() string

func (*WriteError) IsRetryable

func (e *WriteError) IsRetryable() bool

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

type WriteErrorCode

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"
	ErrCodeInternal       WriteErrorCode = "INTERNAL"
)

type WriteMeta

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

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

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

WriteTimeout groups deadline durations for different write categories.