Documentation
¶
Index ¶
- Constants
- Variables
- func BuildUpstreamName(dbbatVersion, username, clientAppName string, maxLen int) string
- func DialUpstream(ctx context.Context, resolver ServerResolver, encryptionKey []byte, ...) (net.Conn, error)
- func EnableClientKeepAlive(conn net.Conn) error
- func IsDDLQuery(sql string) bool
- func IsPasswordChangeQuery(sql string) bool
- func IsWriteQuery(sql string) bool
- func NormalizeSQL(sql string) string
- func ValidateMongoCommand(cmd, dbName string, body bson.Raw, db *store.Server, grant *store.Grant) error
- func ValidateMySQLQuery(sql string, grant *store.Grant) error
- func ValidateOracleQuery(sql string, grant *store.Grant) error
- func ValidateQuery(sql string, grant *store.Grant) error
- type ApprovalDeniedError
- type ApprovalDeps
- type ApprovalEscalator
- type ApprovalGate
- type ApprovalHoldInfo
- type ApprovalStore
- type CountingConn
- type Dialer
- type HoldRequest
- type LimitGuard
- type QuerySink
- func (s *QuerySink) Add(row store.QueryRow) bool
- func (s *QuerySink) AddAll(ctx context.Context, rows []store.QueryRow)
- func (s *QuerySink) Dropped() bool
- func (s *QuerySink) Fail()
- func (s *QuerySink) Flush(ctx context.Context)
- func (s *QuerySink) QueryUID(ctx context.Context) (uuid.UUID, bool)
- func (s *QuerySink) Resolve(queryUID uuid.UUID)
- type RowStore
- type RowWriter
- type ServerResolver
- type StreamPublisher
- type WatchedConn
Constants ¶
const ( // MaxBatchRows is the largest number of rows sent in one INSERT. MaxBatchRows = 1000 // MaxBatchBytes is the largest row payload sent in one INSERT (8 MiB). MaxBatchBytes = 8 << 20 // QueueCapacityRows is the depth of the submit queue, in rows. QueueCapacityRows = 4096 // MaxQueuedBytes bounds the row payload waiting in the queue (32 MiB). // A row cap alone does not bound memory — one wide row can be megabytes — // so the queue enforces both. MaxQueuedBytes = 32 << 20 )
Batch and queue sizing for the row writer.
The batch caps are deliberately large. Batching pays by amortizing the INSERT round-trip, and a bulk insert of 1000 small JSONB rows costs barely more than one of 50 — while at ~1 ms per insert a 50-row cap would ceiling throughput near 50k rows/s, over a minute of pure insert time on a multi-million-row capture.
The byte budget matters as much as the row count: 1000 rows is nothing for a single-column select and tens of megabytes for wide rows, so whichever cap trips first ends the batch.
const ( // ConnectionOpened marks a session that finished authenticating. ConnectionOpened = "opened" // ConnectionClosed marks a session that ended. ConnectionClosed = "closed" )
Connection lifecycle states published on the connections topic.
const ( // ClientKeepAliveIdle is how long a connection may sit idle before the // first probe. ClientKeepAliveIdle = 30 * time.Second // ClientKeepAliveInterval is the gap between probes. ClientKeepAliveInterval = 10 * time.Second // ClientKeepAliveCount is how many unanswered probes declare the peer dead. ClientKeepAliveCount = 3 )
Client TCP keepalive settings. With no approval timeout, a hold ends only on approve, deny, or client disconnect — so "the client went away" has to be something the kernel actually tells us. A hard-killed client or a dead network path never sends a FIN, and without keepalive "until disconnect" silently means "forever", parking an upstream connection and any locks it holds indefinitely.
const DefaultLimitPollInterval = 250 * time.Millisecond
DefaultLimitPollInterval is how often the watchdog re-evaluates limits when no explicit interval is given. Small enough to cut a runaway stream promptly, large enough that the poll cost (two atomic loads + a time compare) is negligible.
Variables ¶
var ( // ErrApprovalAbandoned means the hold ended without a human decision — // the client disconnected, the query was canceled, the grant expired, or // the server drained. Nothing was ever forwarded upstream. It is rendered // distinctly from "denied" everywhere: to whoever finally looks at it, a // query nobody is waiting for anymore is a different thing from a query a // human rejected. ErrApprovalAbandoned = errors.New("approval hold abandoned: client is gone") // (the pending row failed to persist). It fails **closed**: a statement // matching an approval pattern is never forwarded just because the // bookkeeping broke. ErrApprovalUnavailable = errors.New("approval required but the approval system is unavailable") )
Approval hold errors. They travel back through each protocol's existing blocked-query path (the same one ErrDDLBlocked uses), so a denied statement reaches the client as a protocol-native error rather than a dropped socket.
var ( // ErrByteQuotaExceeded indicates the grant's max_bytes_transferred quota // was crossed while data was flowing. ErrByteQuotaExceeded = errors.New("bandwidth quota exceeded for this grant") // ErrGrantExpired indicates the grant's expiry time passed while the // session was still open. ErrGrantExpired = errors.New("grant expired") // ErrGrantRevoked indicates the grant backing the session was revoked // (by an admin, via the API) while the connection was still live. ErrGrantRevoked = errors.New("grant revoked") )
Limit-enforcement errors shared across proxy implementations. They are surfaced both at command boundaries (a new query rejected because the grant is exhausted/expired) and mid-stream (a running query aborted the moment a limit is crossed).
var ( ErrReadOnlyViolation = errors.New("write operations not permitted with read-only access") ErrDDLBlocked = errors.New("DDL operations not permitted: your access grant blocks schema modifications") ErrPasswordChangeBlocked = errors.New("password modification is not allowed through the proxy") ErrOraclePatternBlocked = errors.New("blocked: this Oracle operation is not permitted through the proxy") ErrMySQLPatternBlocked = errors.New("blocked: this MySQL operation is not permitted through the proxy") )
Validation errors shared across proxy implementations.
var ( ErrMongoReadOnly = errors.New("dbbat: grant is read-only") ErrMongoDDLBlocked = errors.New("dbbat: grant blocks DDL operations") ErrMongoCommandBlocked = errors.New("dbbat: command not permitted through dbbat") ErrMongoUnknownCommand = errors.New("dbbat: command not on the proxy allowlist") ErrMongoDatabaseBlocked = errors.New("dbbat: access to this database is not permitted") )
Mongo-specific validation errors (contract §7 surfaces these as the errmsg of an Unauthorized (13) reply).
var ErrApprovalDenied = errors.New("query denied by approver")
ErrApprovalDenied is the sentinel every ApprovalDeniedError matches, so callers can branch with errors.Is without unwrapping.
var ErrBastionNotSSH = errors.New("ssh: via_uid does not reference an ssh server")
ErrBastionNotSSH is returned when a via_uid resolves to a non-ssh row.
var ErrConnParked = errors.New("connection is parked for an approval hold")
ErrConnParked is returned if something tries to read the connection while a hold owns it. It indicates a wiring bug, not a runtime condition.
var ErrNoSSHAuthMethod = errors.New("ssh: bastion has no usable auth method (private key or password)")
ErrNoSSHAuthMethod is returned when a bastion row has neither a private key nor a password to authenticate with.
var ErrSSHHostKeyMismatch = errors.New("ssh: host key mismatch with pinned known_host_key")
ErrSSHHostKeyMismatch is returned when a bastion presents a host key that differs from the TOFU-pinned one recorded on first connect.
var ErrServerViaCycleDial = errors.New("ssh: via_uid chain forms a cycle")
ErrServerViaCycleDial mirrors store.ErrServerViaCycle for the dial path.
Functions ¶
func BuildUpstreamName ¶ added in v0.16.0
BuildUpstreamName composes the canonical dbbat-branded application/program name sent to upstream databases, so a DBA looking at the target's session views (pg_stat_activity.application_name, V$SESSION.PROGRAM, MySQL's process list) can attribute a session to the dbbat user who initiated it.
Format:
dbbat/$version @$username
and, when the client declared an application/program name dbbat was able to intercept:
dbbat/$version @$username for $appName
The result is truncated to fit maxLen, preferring to truncate $appName first so the "dbbat/$version @$username" prefix survives intact. If even the bare prefix exceeds maxLen, the prefix itself is truncated as a last resort. maxLen <= 0 is treated as "no room at all" and returns "".
func DialUpstream ¶ added in v0.17.0
func DialUpstream(ctx context.Context, resolver ServerResolver, encryptionKey []byte, srv *store.Server) (net.Conn, error)
DialUpstream dials srv's host:port using the process-wide pooled dialer. resolver loads the via chain and persists TOFU host keys; encryptionKey decrypts bastion SSH secrets.
func EnableClientKeepAlive ¶ added in v0.20.0
EnableClientKeepAlive turns on TCP keepalive with dbbat's probe settings on the underlying TCP socket. Non-TCP conns (tests, unix sockets) are a no-op. Errors are returned for logging but are never fatal: a proxy that cannot set keepalive is degraded, not broken.
func IsPasswordChangeQuery ¶
IsPasswordChangeQuery checks if a query attempts to modify user/role passwords.
func IsWriteQuery ¶
IsWriteQuery checks if a query is a write operation.
func NormalizeSQL ¶ added in v0.20.0
NormalizeSQL is the canonical normalization applied before pattern matching: a trim, and nothing else.
Note the deliberate divergence from the static validators (IsWriteQuery, IsDDLQuery, IsPasswordChangeQuery), which upper-case before their keyword-prefix checks and are therefore case-insensitive for free. An approval pattern is a full regexp an operator writes and then reads back against the SQL shown in /queries, so rewriting the statement first would make patterns behave differently from what the UI displays.
The cost is that `^DELETE` does not match `delete from …`. Patterns should carry `(?i)` — which the definition form's placeholder and docs/approvals.md both teach — because a pattern that misses is a hold that never happens.
func ValidateMongoCommand ¶ added in v0.16.0
func ValidateMongoCommand(cmd, dbName string, body bson.Raw, db *store.Server, grant *store.Grant) error
ValidateMongoCommand enforces grant controls and the $db policy on a MongoDB command (contract §2). It operates on the command name and the kind-0 body. db is the session's resolved target database; grant carries the controls.
func ValidateMySQLQuery ¶ added in v0.7.0
ValidateMySQLQuery runs shared validation plus MySQL-specific blocked patterns.
func ValidateOracleQuery ¶
ValidateOracleQuery runs shared validation plus Oracle-specific blocked patterns.
Types ¶
type ApprovalDeniedError ¶ added in v0.20.0
ApprovalDeniedError is returned when a human explicitly denied the statement. It carries the approver's reason so each protocol can surface it verbatim to the client.
func (*ApprovalDeniedError) Error ¶ added in v0.20.0
func (e *ApprovalDeniedError) Error() string
func (*ApprovalDeniedError) Is ¶ added in v0.20.0
func (e *ApprovalDeniedError) Is(target error) bool
Is makes errors.Is(err, ErrApprovalDenied) work for the sentinel below.
type ApprovalDeps ¶ added in v0.20.0
type ApprovalDeps struct {
Enabled bool
Store ApprovalStore
Registry *approval.Registry
Broker *events.Broker
Escalator ApprovalEscalator
Logger *slog.Logger
// PollInterval is how often quotas/expiry/revocation are re-evaluated
// while a statement is parked. With no approval timeout, the LimitGuard
// is the one remaining server-side bound, so it must keep running.
PollInterval time.Duration
}
ApprovalDeps are the collaborators every gate in the process shares.
type ApprovalEscalator ¶ added in v0.20.0
type ApprovalEscalator interface {
// Schedule arms the escalation timer for a hold.
Schedule(ctx context.Context, hold ApprovalHoldInfo)
// Resolved cancels a not-yet-fired escalation, or updates the posted
// message in place if it already fired. Because there is no approval
// timeout, a posted message stays actionable indefinitely — a stale
// Approve button that silently no-ops is worse than no button.
Resolved(ctx context.Context, queryUID uuid.UUID, status, byName, reason string)
}
ApprovalEscalator fires the delayed Slack notification for a hold and updates it in place once the hold resolves by any route. Optional: a nil escalator simply means no Slack.
type ApprovalGate ¶ added in v0.20.0
type ApprovalGate struct {
// contains filtered or unexported fields
}
ApprovalGate decides whether a statement needs a human, and parks the session while one is found.
It is constructed once per session: the grant's RE2 patterns are compiled here and reused for every statement, so the per-statement cost of the common (no-pattern) case is a nil check.
func NewApprovalGate ¶ added in v0.20.0
func NewApprovalGate(deps ApprovalDeps, grant *store.Grant, connectionUID uuid.UUID, user *store.User, databaseName string) *ApprovalGate
NewApprovalGate compiles the grant's approval patterns. A gate is returned even when nothing is configured — callers check Active() (or just call Match, which reports false) rather than nil-checking.
Patterns that fail to compile are skipped with a loud log rather than failing the session: they are validated at definition-save time, so a bad one here means data predating validation, and refusing every connection over it would be a worse failure than gating one statement less.
func (*ApprovalGate) Active ¶ added in v0.20.0
func (g *ApprovalGate) Active() bool
Active reports whether this gate can hold anything at all.
func (*ApprovalGate) Hold ¶ added in v0.20.0
func (g *ApprovalGate) Hold(ctx context.Context, req HoldRequest) (uuid.UUID, error)
Hold persists the statement as pending, announces it, and blocks until a human resolves it or the hold dies. It returns the query uid it created (so the caller can complete that row instead of inserting a second one) and nil only on an explicit approval.
Every other exit — deny, disconnect, quota/expiry/revocation, shutdown — returns an error, and no path returns nil without an approval decision whose query uid equals the one this call created. That last clause is the TOCTOU guard: a client able to influence timing must not be able to have somebody else's approval released against its statement.
func (*ApprovalGate) Match ¶ added in v0.20.0
func (g *ApprovalGate) Match(sql string) (string, bool)
Match reports the first pattern the statement matches. Matching runs on the same normalized SQL text the static validators use, so an operator writing a pattern does not have to reason about a second normalization.
type ApprovalHoldInfo ¶ added in v0.20.0
type ApprovalHoldInfo struct {
QueryUID uuid.UUID
ConnectionUID uuid.UUID
UserUID uuid.UUID
Username string
DatabaseName string
SQL string
Pattern string
StartedAt time.Time
}
ApprovalHoldInfo describes a parked statement to the escalator.
type ApprovalStore ¶ added in v0.20.0
type ApprovalStore interface {
CreatePendingQuery(ctx context.Context, query *store.Query, pattern string) (*store.Query, error)
ResolveQueryApproval(ctx context.Context, uid uuid.UUID, status string, resolvedBy *uuid.UUID, reason string) error
NotifyEvent(ctx context.Context, channel string, payload store.EventNotification) error
}
ApprovalStore is the slice of the store the gate needs. An interface rather than *store.Store so protocol tests can drive a hold end-to-end without a database.
type CountingConn ¶ added in v0.10.0
CountingConn wraps a net.Conn and atomically tracks the number of bytes read from and written to it. The two counters live outside the wrapper so a session can share them across multiple wrapped conns (e.g. client and upstream): writes to one direction on one wrapper match reads from the same direction on the other.
Total() is safe to call concurrently with Read/Write — useful for taking per-query snapshots while the proxy is mid-stream.
func NewCountingConn ¶ added in v0.10.0
func NewCountingConn(conn net.Conn, bytesRead, bytesWritten *atomic.Int64) *CountingConn
NewCountingConn wraps conn so Read accumulates into bytesRead and Write accumulates into bytesWritten. Either counter may be nil to disable that direction (rare; the typical caller passes both).
func (*CountingConn) Read ¶ added in v0.10.0
func (c *CountingConn) Read(p []byte) (int, error)
Read implements net.Conn. Successful byte counts are added to the read counter even when the call returns an error (n > 0 with err is a valid outcome on a closing conn — those bytes did cross the wire).
func (*CountingConn) Unwrap ¶ added in v0.20.0
func (c *CountingConn) Unwrap() net.Conn
Unwrap exposes the wrapped conn so helpers that need the raw socket (EnableClientKeepAlive) can walk down the wrapper chain.
type Dialer ¶ added in v0.17.0
type Dialer struct {
// contains filtered or unexported fields
}
Dialer opens upstream connections, tunneling through SSH bastions when a server row's ViaUID is set. It pools one *ssh.Client per bastion (keyed by server UID) so that N concurrent proxy sessions multiplex over a single SSH connection; a dead client is transparently reconnected.
func NewDialer ¶ added in v0.17.0
func NewDialer() *Dialer
NewDialer builds an empty Dialer with its own bastion pool.
func (*Dialer) Close ¶ added in v0.18.0
func (d *Dialer) Close()
Close tears down every pooled bastion client. Used by short-lived dialers (connectivity checks) so a probe does not leak an SSH connection.
func (*Dialer) ConnectBastion ¶ added in v0.18.0
func (d *Dialer) ConnectBastion( ctx context.Context, resolver ServerResolver, encryptionKey []byte, uid uuid.UUID, ) (*ssh.Client, error)
ConnectBastion dials (or reuses a pooled connection to) the SSH bastion row identified by uid, completing the handshake and — on first connect — pinning the presented host key via resolver.SetKnownHostKey.
It exists so a connectivity check can validate a `protocol: ssh` row on its own, with no database target behind it. Callers that want to force a real dial (rather than reuse a pooled client) must use a fresh Dialer.
func (*Dialer) DialUpstream ¶ added in v0.17.0
func (d *Dialer) DialUpstream(ctx context.Context, resolver ServerResolver, encryptionKey []byte, srv *store.Server) (net.Conn, error)
DialUpstream dials srv's host:port directly, or through srv.ViaUID's SSH bastion chain when set (recursing for multi-hop jump hosts).
type HoldRequest ¶ added in v0.20.0
type HoldRequest struct {
// SQL is the statement text, as it will be persisted and shown.
SQL string
// Params are the bind parameters, known at Execute time (not at Parse
// time — which is exactly why the PostgreSQL gate hooks Execute).
Params *store.QueryParameters
// Pattern is the pattern that matched, recorded on the row.
Pattern string
// StartedAt is when the statement arrived.
StartedAt time.Time
// ClientGone fires when the parked client's socket dies. This is the
// sole liveness bound on a hold: without it, "until disconnect" means
// forever.
ClientGone <-chan struct{}
// Guard keeps quotas, expiry and revocation running while parked.
Guard *LimitGuard
// OnPending, when set, is called with the pending row's uid the instant
// it exists — before the session blocks. Protocols use it to publish the
// uid to their out-of-band cancellation path (PostgreSQL CancelRequest,
// MySQL KILL QUERY, Mongo killOperations), which must be able to end a
// hold that has not returned yet.
OnPending func(queryUID uuid.UUID)
}
HoldRequest is one parked statement.
type LimitGuard ¶ added in v0.16.0
type LimitGuard struct {
// contains filtered or unexported fields
}
LimitGuard evaluates a grant's time-window and bandwidth limits against the live wire-byte counters. It is designed to be called on the data path: Check() performs at most two atomic loads and a wall-clock comparison, with no allocation and no locking.
A guard built from a nil grant (or a grant with no limits) never trips, so callers can construct one unconditionally.
func NewLimitGuard ¶ added in v0.16.0
func NewLimitGuard(grant *store.Grant, from, to *atomic.Int64) *LimitGuard
NewLimitGuard builds a guard for grant, reading live traffic from the two atomic counters (either may be nil). grant may be nil — the resulting guard enforces nothing.
func (*LimitGuard) Check ¶ added in v0.16.0
func (g *LimitGuard) Check() error
Check reports the first limit that has been crossed, or nil if the grant is still within bounds. Bandwidth is checked before expiry so the "gigabytes in seconds" case is attributed to the byte quota, but either is a valid abort reason.
func (*LimitGuard) Watch ¶ added in v0.16.0
Watch polls Check on a ticker until a limit is crossed or ctx is canceled. On the first violation it invokes onViolation with the offending error and returns; onViolation is never called more than once. interval <= 0 falls back to DefaultLimitPollInterval.
Watch is the guaranteed, protocol-agnostic enforcement path: it fires even when a query is blocked producing no traffic (idle expiry) and even for protocols whose client library owns the wire (MySQL). onViolation typically force-closes the client and upstream conns to tear the session down.
func (*LimitGuard) WithRevocation ¶ added in v0.16.0
func (g *LimitGuard) WithRevocation(revoked *atomic.Bool) *LimitGuard
WithRevocation attaches the session's shared revocation flag to the guard so Check/Watch also trip when the grant is revoked mid-session. Returns the guard for fluent construction. A nil flag is a no-op (nothing to watch), keeping the plain NewLimitGuard signature stable for callers/tests that don't track revocation.
type QuerySink ¶ added in v0.20.0
type QuerySink struct {
// contains filtered or unexported fields
}
QuerySink is one query's handle on the shared writer. It carries the parent query uid (once known) and records whether the capture lost rows.
A nil sink is inert: Add reports the row as not accepted, everything else is a no-op. That is what a session with no store gets.
func (*QuerySink) Add ¶ added in v0.20.0
Add queues one captured row. It never blocks: on a full queue (or an exhausted byte budget) the row is dropped, the capture is marked degraded, and false is returned. This is the entry point for capture paths that run inline with forwarding rows to the client.
func (*QuerySink) AddAll ¶ added in v0.20.0
AddAll queues a whole result set, waiting for queue space rather than dropping. It is only for callers that already hold every row in memory and run off the data path (MySQL, MongoDB, PostgreSQL COPY): those rows are already resident, so queueing them costs no extra memory, and dropping them would lose a capture that used to be stored whole.
The sink must already be resolved — the drain goroutine cannot make progress on rows whose parent query record is still unknown while the caller holds it.
func (*QuerySink) Dropped ¶ added in v0.20.0
Dropped reports whether any row was lost — queue full, batch insert failed, or no parent query record. Read it after Flush so the answer covers the whole capture.
func (*QuerySink) Fail ¶ added in v0.20.0
func (s *QuerySink) Fail()
Fail reports that the query record could not be created, so the sink's rows have no parent to hang from. Queued rows are discarded and the capture is marked as having dropped rows.
func (*QuerySink) Flush ¶ added in v0.20.0
Flush blocks until every row submitted so far has been inserted (or definitively lost). It is the barrier a query needs before it is marked complete: without it the UI would show a finished query with rows still arriving.
Call it from the completion goroutine, never from the capture path.
func (*QuerySink) QueryUID ¶ added in v0.20.0
QueryUID reports the uid of the query record this sink's rows hang from, blocking until it is known. It reports false when there is no sink, when the record could not be created, or when nothing resolved it in time — in which case the caller owns creating the record itself.
type RowStore ¶ added in v0.20.0
type RowStore interface {
StoreQueryRows(ctx context.Context, rows []store.PendingQueryRow) error
}
RowStore is the slice of the store the row writer needs.
type RowWriter ¶ added in v0.20.0
type RowWriter struct {
// contains filtered or unexported fields
}
RowWriter persists captured result rows in batches, off the proxy's data path. One writer is shared by the whole process: its batches span protocols, sessions and queries, so a busy proxy running many small result sets issues one INSERT per ~1000 rows overall rather than one per query.
The queue is a bounded channel drained opportunistically: the drain goroutine blocks for the first row, then takes whatever is already queued until a cap trips. Batches therefore size themselves to load — an idle producer gets batches of one and minimal latency, a fast producer gets full batches — with no timer to tune.
Submitting from a capture path is non-blocking and drops on a full queue (see QuerySink.Add). The capture path runs inline with forwarding rows to the client, so a slow moment in dbbat's own storage must never become a stall on the customer's query.
A nil *RowWriter is usable: every method is a no-op and NewSink returns a nil sink, which is likewise inert. That keeps the protocol call sites free of nil checks when there is no store to write to.
func NewRowWriter ¶ added in v0.20.0
NewRowWriter starts a writer against st. A nil store yields a nil writer, which is inert — callers with no store keep working unchanged.
func (*RowWriter) Close ¶ added in v0.20.0
Close drains what is queued and stops the writer. It is safe to call twice.
func (*RowWriter) NewSink ¶ added in v0.20.0
NewSink returns a sink for one query whose record does not exist yet. Rows may be submitted immediately; they are held in the queue and only inserted once Resolve supplies the parent query uid, which is what keeps the query_rows -> queries foreign key satisfied without the producer ever blocking on the parent INSERT.
type ServerResolver ¶ added in v0.17.0
type ServerResolver interface {
GetServerByUID(ctx context.Context, uid uuid.UUID) (*store.Server, error)
SetKnownHostKey(ctx context.Context, uid uuid.UUID, hostKey string) error
}
ServerResolver resolves server rows and persists TOFU host keys. Satisfied by *store.Store; an interface so the dialer can be unit-tested with a fake.
type StreamPublisher ¶ added in v0.20.0
type StreamPublisher struct {
// contains filtered or unexported fields
}
StreamPublisher publishes a session's observable activity — every query on the connection, plus connection open/close — to the live event stream.
It is separate from ApprovalGate on purpose: streaming is unconditional and harmless, approval holds are opt-in and block a database connection. They share nothing but the connection identity.
Every method is nil-safe and non-blocking. A broken stream must never break a database connection, which is why nothing here returns an error.
func NewStreamPublisher ¶ added in v0.20.0
func NewStreamPublisher( deps ApprovalDeps, connectionUID uuid.UUID, user *store.User, databaseName string, ) *StreamPublisher
NewStreamPublisher builds a publisher for one session. A nil broker yields a publisher whose methods are no-ops.
func (*StreamPublisher) Connection ¶ added in v0.20.0
func (p *StreamPublisher) Connection(ctx context.Context, state string)
Connection announces a connection lifecycle change on the admin-only connections topic.
func (*StreamPublisher) Query ¶ added in v0.20.0
func (p *StreamPublisher) Query(queryUID uuid.UUID, q *store.Query)
Query announces one executed statement on connection/<uid>/queries.
There is deliberately no global all-queries topic: a busy proxy issues thousands of statements a second, each carrying full SQL text, and firehosing that would be both a throughput problem and a data-exposure one. Watching is per-connection and opt-in.
type WatchedConn ¶ added in v0.20.0
WatchedConn is a net.Conn wrapper that can be "parked": while parked, a background goroutine keeps reading the socket so a client FIN is noticed immediately instead of sitting unread until the session resumes.
This is load-bearing rather than decorative. During an approval hold the session goroutine is blocked on a human, so nothing is reading the client socket; without an active read-watch a disconnected client would leave the statement parked forever.
Bytes a pipelining client sent while parked are *not* dropped and *not* interpreted mid-hold: they are queued and replayed, in stream order, the moment the session resumes reading.
Placement matters: WatchedConn must sit *below* any TLS layer, so it parks on raw TLS records it never has to decrypt.
func NewWatchedConn ¶ added in v0.20.0
func NewWatchedConn(conn net.Conn) *WatchedConn
NewWatchedConn wraps conn.
func (*WatchedConn) Buffered ¶ added in v0.20.0
func (w *WatchedConn) Buffered() int
Buffered reports how many replay bytes are queued. Test/telemetry helper.
func (*WatchedConn) Park ¶ added in v0.20.0
func (w *WatchedConn) Park() <-chan struct{}
Park starts the read-watch. The returned channel is closed when the client goes away (EOF, reset, or any other terminal read error). Unpark must be called before the session resumes reading.
func (*WatchedConn) Read ¶ added in v0.20.0
func (w *WatchedConn) Read(p []byte) (int, error)
Read serves any bytes captured while parked before touching the socket, so the byte stream the session sees is exactly the byte stream the client sent.
func (*WatchedConn) Unpark ¶ added in v0.20.0
func (w *WatchedConn) Unpark()
Unpark stops the read-watch and hands the socket back to the session. Any bytes the watcher captured stay queued for replay.
func (*WatchedConn) Unwrap ¶ added in v0.20.0
func (w *WatchedConn) Unwrap() net.Conn
Unwrap exposes the wrapped conn so helpers like EnableClientKeepAlive can reach the TCP socket underneath.
Source Files