joinexec

package
v0.14.2 Latest Latest
Warning

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

 Go to latest Published: Oct 5, 2026 License: Apache-2.0 Imports: 17 Imported by: 1

Documentation

Overview

Package joinexec is the security boundary for joined reads.

DALgo evaluates a joined query by reading each source with a single-source query and combining the rows itself. Every one of those reads passes through a leaf made by a Guard, and the leaf is the one place that

  • authorises the collection before it is read (a denied collection is never read, and never has its schema looked up),
  • refuses anything that is not a plain query over one collection,
  • counts rows and encoded bytes against the request budget while streaming, and stops with a BudgetError when the budget is exceeded,
  • records per-source statistics for the response.

One Guard serves one request. Create it with NewGuard, take its context from Guard.Context, make a leaf per source with Guard.Leaf, hand the leaves to DALgo (for example through a dal.DatabaseResolver), and answer the caller with Guard.Collect, which drains the reader DALgo returns, closes it and returns either every row or no rows and a classified error. A caller that reads the reader itself must read it to the end, close it, and only then pass the error it ended with (or nil) to Guard.Classify; rows read before an error are not an answer, because DALgo's streaming join delivers its rows before a budget error arrives.

Index

Constants

View Source 
const (
	MaxInMemoryJoinRows  = dalgoJoinRows
	MaxInMemoryJoinBytes = dalgoJoinBytes
	MaxInMemoryGroups    = dalgoAggregationGroups
)

The bounds of the in-memory route, which the server states to a client in its discovery document and in the answer to a request that reaches one: the most rows and bytes a join holds, and the most groups an aggregation keeps.

View Source 
const (
	RouteInMemory = "in-memory"
	RouteDatabase = "database"
)

Execution routes a BudgetError can name. They are the same labels the HTTP response reports as execution.route.

View Source 
const (
	BudgetSourceRows  = "source_rows"
	BudgetSourceBytes = "source_bytes"

	BudgetJoinRows                 = "join_rows"
	BudgetJoinResultRows           = "join_result_rows"
	BudgetJoinFetchedRows          = "join_fetched_rows"
	BudgetJoinRetainedBytes        = "join_retained_bytes"
	BudgetJoinScan                 = "join_scan"
	BudgetJoinCandidateEvaluations = "join_candidate_evaluations"

	BudgetAggregationGroups         = "aggregation_groups"
	BudgetAggregationStates         = "aggregation_states"
	BudgetAggregationBytes          = "aggregation_bytes"
	BudgetAggregationDistinctValues = "aggregation_distinct_values"
	BudgetAggregationTotalDistinct  = "aggregation_total_distinct_values"
)

Budget names carried by BudgetError.Name. The source_* names are this package's own request budget; the join_* and aggregation_* names are the bounds DALgo enforces itself (see MapDalgoError).

View Source 
const (
	// MaxResultRows is the most rows a result may have.
	MaxResultRows = 1000
	// MaxResultBytes is the most JSON-encoded bytes a result may have.
	MaxResultBytes = 8 << 20
)

The result of one request is bounded, whatever the sources hold: a result over either bound is refused with a BudgetError, never cut short.

View Source 
const (
	BudgetResponseRows  = "response_rows"
	BudgetResponseBytes = "response_bytes"
)

Budget names of the result bound, carried by BudgetError.Name beside the ones of errors.go. They bound what is answered, not what is read.

Variables

View Source 
var (
	// ErrNotSingleSource is returned for a read that is not a plain structured
	// query over exactly one collection: a join, a derived source, a query
	// holding a subquery, or a text query. Only single-source reads reach a
	// source, because only those can be authorised one collection at a time.
	ErrNotSingleSource = errors.New("joinexec: a source read must be a plain query over one collection")

	// ErrUnsupportedSource is returned for a collection the request profile
	// does not allow: schema-qualified or nested under a parent key. Their
	// names cannot be matched against collection grants or policies.
	ErrUnsupportedSource = errors.New("joinexec: schema-qualified and nested collections are not supported")

	// ErrRecordsetUnsupported is returned by the recordset read path, which is
	// not guarded and therefore not offered.
	ErrRecordsetUnsupported = errors.New("joinexec: recordset readers are not supported")

	// ErrNoExecutor is returned when a Source has no executor to read through.
	ErrNoExecutor = errors.New("joinexec: source has no executor")

	// ErrRowNotEncodable is returned when a row cannot be encoded to measure it
	// against the byte budget. The read fails closed.
	ErrRowNotEncodable = errors.New("joinexec: source row cannot be encoded")
)
View Source 
var (
	// ErrInvalidDocument is returned when the document is not one this package
	// executes: a shape the walk does not know, a name that is not plain, or a
	// profile that does not describe the query. The document was not run, and no
	// mount was resolved for it.
	ErrInvalidDocument = errors.New("joinexec: invalid document")

	// ErrProfileMismatch is returned, wrapped in ErrInvalidDocument as well, when
	// the Profile handed to Execute does not describe the query the walk found:
	// another set of sources or another answer about subqueries. It tells a
	// disagreement between the classifier that built the profile and the walk (a
	// defect of the server) from a document that is merely invalid.
	ErrProfileMismatch = errors.New("joinexec: the profile does not match the query")

	// ErrSourceWithoutDatabase is returned when a source names no database and
	// there is no single default one to read it from: the request has no default
	// database, or the document reads several databases and so must name the
	// database of every source.
	ErrSourceWithoutDatabase = errors.New("joinexec: a source names no database")

	// ErrScanOnProtectedSource is returned for a scan clause on a source with
	// access policies. DALgo applies the scan's limit and order to the source's
	// own read, where the policy filter could run after the limit, so the clause
	// is refused until DALgo applies policies before scan bounds (dalgo pull
	// request 197).
	ErrScanOnProtectedSource = errors.New("joinexec: scan clauses are not supported on a source with access policies")

	// ErrReadTxSkipped is returned when a source's ReadTx returned without
	// running the function it was given, which would otherwise read as an empty
	// result.
	ErrReadTxSkipped = errors.New("joinexec: the read transaction did not run the query")

	// ErrNoReader is returned when an executor returned neither a reader nor an
	// error.
	ErrNoReader = errors.New("joinexec: the executor returned no reader")

	// ErrReadTruncated is returned when the reader of the database route failed
	// with an error that wraps io.EOF. A reader's end is a bare io.EOF; an error
	// that merely wraps it is a failure part way (a dropped connection reports
	// itself this way), and answering it as the end would return the rows read
	// so far as a complete result. The original error is in the message, not in
	// the chain, so that nothing reads it as the end of the stream.
	ErrReadTruncated = errors.New("joinexec: the read ended with a failure that looks like the end of the stream")

	// ErrRegistryMismatch is returned when the registry answers a database id
	// with a source of another id. The leaf authorises a read under the source's
	// own id, so such a source would be authorised under the wrong name.
	ErrRegistryMismatch = errors.New("joinexec: the registry returned a source for another database")
)

Functions

func IsJoinPlanRefusal added in v0.12.3 

func IsJoinPlanRefusal(message string) bool

IsJoinPlanRefusal reports whether message, the Message of a join_plan error of DALgo's join, is the refusal of a document a caller made and can change: a wildcard over a source with no field list, a cursor, an array that is not one, a null test of nothing, an operator or a type of expression or condition the join does not evaluate. Any other message of the category is a failed read, a fault of encoding or a bound.

func MapDalgoError 

func MapDalgoError(err error, route string) error

MapDalgoError returns a *BudgetError when err is, or wraps, one of DALgo's own bound errors, and err unchanged otherwise (including nil). route is the route the engine ran on (RouteInMemory or RouteDatabase). An error that is already a *BudgetError or *SourceDeniedError is returned as is.

DALgo reports a leaf's error as text inside its own error (and, since dalgo v0.89.6, as the cause of a join_plan diagnostic); use Guard.Classify to recover a leaf's BudgetError, SourceDeniedError or SourceError as well.

Types

type Admit added in v0.11.14 

type Admit func(ctx context.Context, route string) (release func(), ok bool)

Admit takes a slot for a request that is about to read, on the route Execute chose for it (RouteDatabase or RouteInMemory). It returns the function that gives the slot back and true, or false when there is no slot, in which case the release function is not used. Execute calls the release function when the request ends, whatever way it ends; a nil release function is treated as one that does nothing.

type Authorize 

type Authorize func(database, collection string) bool

Authorize reports whether the caller may read the collection of the database. It is called before every read; a nil Authorize denies everything.

type BudgetError 

type BudgetError struct {
	// Name is one of the Budget* constants.
	Name string
	// Limit is the bound that was exceeded. It is 0 when DALgo's message does
	// not identify which of several limits applied.
	Limit int64
	// Route is RouteInMemory or RouteDatabase.
	Route string
	// Path locates the bound: "database.collection" for a source read, or the
	// join tree path DALgo reports (for example "from.joins[0]"). It can be
	// empty when DALgo gives no path.
	Path string
}

BudgetError reports that a request exceeded a resource bound. It names the bound and its limit, never the observed figure.

A BudgetError ends the request, but rows may already have been read when it arrives: DALgo's streaming join (a flat equality join without ORDER BY) returns its reader first and the error comes from a later Next. Only a caller that reads to the end before answering returns no rows; Guard.Collect does exactly that.

func (*BudgetError) Error 

func (e *BudgetError) Error() string

type CapacityError added in v0.11.14 

type CapacityError struct {
	// Route is RouteInMemory or RouteDatabase.
	Route string
}

CapacityError reports that no slot was free for a request on its route: the server is running as many queries of that kind as it allows. Nothing was read.

func (*CapacityError) Error added in v0.11.14 

func (e *CapacityError) Error() string

type EngineNotJoinableError added in v0.11.11 

type EngineNotJoinableError struct {
	Database string
	Engine   string
}

EngineNotJoinableError reports a database whose storage engine is not in the configured set of engines that may take part in relational queries.

func (*EngineNotJoinableError) Error added in v0.11.11 

func (e *EngineNotJoinableError) Error() string

type EngineNotQueryableError added in v0.11.11 

type EngineNotQueryableError struct {
	Database string
	Engine   string
}

EngineNotQueryableError reports a database whose storage engine cannot run structured queries at all. Callers answer it as the engine guard of the single-collection path does (HTTP 501 query_unsupported). It takes precedence over EngineNotJoinableError, and no operator setting lifts it.

func (*EngineNotQueryableError) Error added in v0.11.11 

func (e *EngineNotQueryableError) Error() string

type Execution added in v0.11.11 

type Execution struct {
	// Route is RouteDatabase or RouteInMemory. It names the path, not where every
	// filter ran: on RouteInMemory DALgo evaluates the document above guarded
	// single-collection reads, and a document with one source that names its
	// database and has no subquery, null test, aggregation or scan clause is read
	// with one such query that carries its WHERE, ORDER BY and LIMIT, which the
	// mount's own executor runs. Every other in-memory document (one with a join,
	// GROUP BY, HAVING, an aggregate, a scan clause, or a source that names no
	// database) is read with plain scans of each collection and filtered above
	// them.
	Route        string            `json:"route"`
	ElapsedMs    int64             `json:"elapsedMs"`
	RowsReturned int               `json:"rowsReturned"`
	Sources      []ExecutionSource `json:"sources"`
}

Execution describes how a request ran. It is the execution block of the response and never reports how many rows a source holds or was read for, only how many rows it delivered (and not even that for a protected source).

type ExecutionSource added in v0.11.11 

type ExecutionSource struct {
	Database   string `json:"database"`
	Collection string `json:"collection"`
	// Rows is how many rows the source delivered. It is nil for a source with
	// access policies, whose size a caller may not learn, and on the database
	// route, where the database ran the whole document and reports no more.
	Rows *int `json:"rows,omitempty"`
	// ElapsedMs is the time the source's readers were open, nil on the database
	// route.
	ElapsedMs *int64 `json:"elapsedMs,omitempty"`
}

ExecutionSource describes the reads of one collection.

type Guard 

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

Guard holds the budget, statistics and first failure of one request. It is safe for concurrent use by the leaves it made.

func NewGuard 

func NewGuard(authorize Authorize, limits Limits) *Guard

NewGuard returns the Guard for one request. Zero fields of limits take their defaults (see Limits).

func (*Guard) Classify 

func (g *Guard) Classify(err error, route string) error

Classify turns what a request ended with into the error to answer with. err is the error DALgo returned, or the one a drained reader ended with, or nil.

DALgo reports a leaf's error as text inside its own error (and reads an error that wraps io.EOF as the end of a stream), so the recorded failure wins when there is one, even for a nil err: a request that recorded a failure never ends well. (Since dalgo v0.89.6 the diagnostic also carries the leaf's error as its cause; the answer is still the recorded error, never DALgo's wrapper.) Otherwise DALgo's own bounds map to BudgetError (see MapDalgoError) and any other error is returned unchanged. Call it once, after the reader was drained and closed.

func (*Guard) Collect 

func (g *Guard) Collect(ctx context.Context, reader dal.RecordsReader, route string) ([]record.Record, error)

Collect drains reader, closes it and returns every row, or no rows and the classified error when the read failed in any way: an error from the reader, a failure the guard recorded even if the reader ended cleanly, a close error, or the end of ctx. It is the one way to answer a request that cannot return a partial result. reader must be non-nil, so a caller whose DALgo call returned an error answers with Classify instead.

func (*Guard) Context 

func (g *Guard) Context(parent context.Context) (context.Context, context.CancelFunc)

Context returns parent bounded by the request timeout. The caller must call the returned cancel function.

func (*Guard) Err 

func (g *Guard) Err() error

Err returns the first failure a leaf recorded (a SourceDeniedError, a BudgetError, a SourceError carrying the source's own error, or a refusal such as ErrNotSingleSource), or nil. Once set, every further read through any leaf of this Guard fails with it.

func (*Guard) Leaf 

func (g *Guard) Leaf(src Source) dal.QueryExecutor

Leaf returns the guarded executor for one source. It implements dal.QueryExecutor and dal.JoinFieldsProvider.

func (*Guard) Stats 

func (g *Guard) Stats() []SourceStats

Stats returns the per-collection statistics in the order collections were first read. It is complete only after every reader has been closed; Collect closes the reader it drains.

type Limits 

type Limits struct {
	// MaxSourceRows caps the rows read from all sources together.
	MaxSourceRows int
	// MaxSourceBytes caps the JSON-encoded bytes read from all sources together.
	MaxSourceBytes int64
	// Timeout bounds the whole request; see Guard.Context.
	Timeout time.Duration
}

Limits bounds one request. A zero or negative field takes its default from DefaultLimits, so an unconfigured server is bounded rather than unbounded.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits are the request bounds for a 512 MiB instance.

type Option added in v0.11.11 

type Option func(*config)

Option configures Execute.

func WithAdmission added in v0.11.14 

func WithAdmission(admit Admit) Option

WithAdmission makes Execute ask admit for a slot once it knows the route and before it reads anything, and hold the slot until it returns. A request that fails earlier (a document that is not valid, a source the caller may not read, an unknown database, an engine that is refused) holds no slot and asks for none. When admit has no slot Execute returns a *CapacityError and has read nothing. The wait for a slot is part of the elapsed time Execute reports, and not of the request timeout, which starts when the read does. A nil admit admits every request.

func WithJoinEngines added in v0.11.11 

func WithJoinEngines(engines ...string) Option

WithJoinEngines sets the storage engines whose databases may take part in a relational query. The default is sqlite and ingitdb, as the server's default query limits; an empty list keeps the default. An engine outside the set is refused with an EngineNotJoinableError.

func WithNativeEngines added in v0.11.11 

func WithNativeEngines(engines ...string) Option

WithNativeEngines sets the storage engines that run a whole document themselves (the database route). The default is sqlite; an empty list keeps the default. An engine listed here must also be in the join set.

type Profile added in v0.11.11 

type Profile struct {
	// Sources lists every collection the document reads, including reads inside
	// derived sources and subquery expressions; a collection read twice is
	// listed twice. Database is empty for a source that names none.
	Sources []ProfileSource
	// HasSubquery reports a derived source, scalar subquery, EXISTS predicate
	// or query-valued comparison operand anywhere in the document.
	HasSubquery bool
}

Profile is the classification of the document, as the profile classifier of pkg/core reports it. It is a claim, not evidence: Execute walks the query itself and refuses a profile that does not describe it (ErrInvalidDocument).

type ProfileSource added in v0.11.11 

type ProfileSource struct {
	Database   string
	Collection string
}

ProfileSource is one collection read of a Profile.

type Registry added in v0.11.11 

type Registry interface {
	// Lookup returns the source registered under database, or false when there
	// is none.
	Lookup(database string) (Source, bool)
}

Registry resolves a database id to the source registered under it. A server that keeps a mount alive for the length of a request takes that hold in Lookup and lets go of it when the request ends.

type Result added in v0.11.11 

type Result struct {
	Records []record.Record
	// Columns names the columns of the records in the order they were selected.
	Columns   []string
	Execution Execution
}

Result is the answer to one request.

func Execute added in v0.11.11 

func Execute(ctx context.Context, query dal.StructuredQuery, profile Profile, defaultDatabase string, registry Registry, authorize Authorize, limits Limits, opts ...Option) (Result, error)

Execute runs a relational document and returns every row of its result, or no rows and an error. It is the one place that reads a joined result, and it reads with Guard.Collect, so a request that ran out of budget or lost a source part way is never answered with the rows read before the failure.

defaultDatabase is the database of a source that names none; it is empty on an endpoint that serves several databases, where every source must name one. A source with a database of its own is read from that database whatever the default is: an endpoint that serves one database must refuse a document that names another itself.

Execute works in this order, and a request that fails at one step has done nothing of the steps after it:

  1. Walk the document, check every name in it and check the profile against the walk (ErrInvalidDocument).
  2. Settle the database of every source (ErrSourceWithoutDatabase).
  3. Authorise every source the walk found (SourceDeniedError), before any mount is resolved, so a caller learns nothing about databases it may not read: 403 before 404.
  4. Resolve each database once (UnknownDatabaseError).
  5. Refuse an engine that cannot be queried (EngineNotQueryableError), for every source, and then one outside the join set (EngineNotJoinableError).
  6. Refuse a scan clause on a source with access policies (ErrScanOnProtectedSource).
  7. Choose the route, take a slot for it when WithAdmission is set (CapacityError), and run. The retry in memory takes a slot of its own.

The database route is chosen when every source is in one database, that database's engine is native, it has no access policies, the document has no subquery and no null test, and no source has a scan clause (no engine adapter applies a scan bound or compiles a null test). The whole document then runs in one read transaction of the database. A join with no aggregation that the database answers with an enforcement-unsupported denial (its adapter cannot compile the document, and says so before any row is read) is read again in memory, after the first slot is given back; see request.retryInMemory. Every other document runs in memory: DALgo reads each source with a single-collection query through a guarded leaf and joins, filters and aggregates above the leaves, so a source with access policies is only ever read through its policy-checked executor and only rows the caller may read reach a join or an aggregate. See Execution.Route for what the in-memory label does and does not say about where a filter ran.

A Source must answer CanQuery, which says whether its engine can run structured queries at all (core.Database has it). A Source without the method is treated as one that cannot be queried, so the gate does not depend on the method set of whatever type a registry returns. The check comes before the join-set check and no join-set setting lifts it.

A Source may also answer CheckRead(dal.Query) error (core.Database has it): the checks a read through the source runs before its driver is reached, asked without a transaction. The database route asks it before ReadTx, so a document the source refuses begins no transaction; a Source without the method is refused only inside its transaction, by whatever guards its executor.

The walk checks every name of a document (fields, aliases, qualifiers, parameters, the result names of scalar subqueries and collections) by the strict rules of pkg/core, the arithmetic operators and aggregate names by the classifier's lists, refuses a query whose columns carry one output name twice (the later column would replace the earlier one in a row), refuses a document that still holds a parameter (a parameter is bound before a document runs, and DALgo's join evaluates none), and refuses a document whose conditions and expressions nest more than 64 levels. A field name must pass the strict rule (core.ValidateFieldName) whatever the route and the engine. The classifier of pkg/core applies a wider quoted-name rule to the field names of a relational document, so a name such as "zip code" classifies and Execute then refuses it with ErrInvalidDocument. What only the classifier checks, and Execute does not repeat, is the format of a database id, the limit and offset bounds, money, cursors, the join types, the number of sources, how many levels of subquery nest (four), the refusal of every scan clause (Execute accepts one on a source without access policies and reads such a document in memory), and the sixteen levels at which its name check stops. A caller must therefore hand Execute a document, and a profile, that passed the classifier.

opts, limits and the ordering of results are as documented on Option, Limits, MaxResultRows and MaxResultBytes.

type Source 

type Source interface {
	// ID is the database id the source is registered under. It is the database
	// name Authorize receives and the name a query's collections must carry.
	ID() string
	// Engine is the storage engine name, for routing decisions made by the
	// caller.
	Engine() string
	// HasAccessPolicies reports whether Executor is wrapped in access
	// policies. Statistics of such a source are marked Protected.
	HasAccessPolicies() bool
	// Executor reads the source with the request principal applied. Every read
	// of the in-memory route goes through it.
	Executor() dal.QueryExecutor
	// ReadTx runs fn in one read transaction on the source, for the database
	// route. Reads made through the executor passed to fn are not guarded by a
	// leaf; the caller authorises such a query before running it.
	ReadTx(ctx context.Context, fn func(dal.QueryExecutor) error) error
}

Source is one registered database as the join executor sees it. The server's mount type implements it; this package depends on nothing else of the server.

type SourceDeniedError 

type SourceDeniedError struct {
	Database   string
	Collection string
}

SourceDeniedError reports a read of a collection the caller may not read. The collection was never read.

func (*SourceDeniedError) Error 

func (e *SourceDeniedError) Error() string

type SourceError 

type SourceError struct {
	Database   string
	Collection string
	Err        error
}

SourceError reports that a source failed while it was being read: its executor returned an error, a reader failed part way, or the request context ended during a read. Err is the source's own error with its chain intact, so errors.Is finds an access denial, a deadline or a backend sentinel through it. DALgo rewraps such an error as text; the guard records this one so that Guard.Classify can return it instead.

func (*SourceError) Error 

func (e *SourceError) Error() string

func (*SourceError) Unwrap 

func (e *SourceError) Unwrap() error

type SourceStats 

type SourceStats struct {
	Database   string
	Collection string
	// Rows is the number of rows delivered to the executor.
	Rows int
	// Elapsed is the time readers of this collection were open, summed over
	// reads.
	Elapsed time.Duration
	// Protected is true when the source has access policies. Callers should
	// not report Rows for a protected source.
	Protected bool
}

SourceStats describes the reads of one collection during a request.

type UnknownDatabaseError added in v0.11.11 

type UnknownDatabaseError struct {
	Database string
}

UnknownDatabaseError reports a database id no source is registered under. It is returned only after every source of the document was authorised, so it does not tell a caller who may not read a database whether it exists.

func (*UnknownDatabaseError) Error added in v0.11.11 

func (e *UnknownDatabaseError) Error() string

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.