Struct QueryRuntime

Source

pub struct QueryRuntime<T: Tracer = NoopTracer> { /* private fields */ }

Expand description

The query runtime manages query execution, caching, and dependency tracking.

This is cheap to clone - all data is behind Arc.

§Type Parameter

T: Tracer - The tracer type for observability. Use NoopTracer (default) for zero-cost when tracing is not needed.

§Example

use query_flow::{query, Db, NoopTracer, QueryError, QueryRuntime};

#[query]
fn my_query(db: &impl Db, x: i32) -> Result<i32, QueryError> {
    let _ = db;
    Ok(x * 2)
}

// Without tracing (default)
let runtime = QueryRuntime::new();

// With tracing (see the `tracer` module for writing a custom tracer)
let tracer = NoopTracer;
let runtime = QueryRuntime::with_tracer(tracer);

// Sync query execution
let result = runtime.query(MyQuery::new(21)).unwrap();
assert_eq!(*result, 42);

Implementations§

Source §

impl<T: Tracer> QueryRuntime<T>

Source

pub fn tracer(&self) -> &T

Get a reference to the tracer.

Source §

impl QueryRuntime<NoopTracer>

Source

pub fn new() -> Self

Create a new query runtime with default settings.

Source

pub fn builder() -> QueryRuntimeBuilder<NoopTracer>

Create a builder for customizing the runtime.

§Example

use std::fmt;

use query_flow::QueryRuntime;

#[derive(Debug, PartialEq)]
struct MyError(u32);

impl fmt::Display for MyError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "my error {}", self.0)
    }
}

impl std::error::Error for MyError {}

let runtime = QueryRuntime::builder()
    .error_comparator(|a, b| {
        // Custom error comparison logic
        match (a.downcast_ref::<MyError>(), b.downcast_ref::<MyError>()) {
            (Some(a), Some(b)) => a == b,
            _ => false,
        }
    })
    .build();

Source §

impl<T: Tracer> QueryRuntime<T>

Source

pub fn with_tracer(tracer: T) -> Self

Create a new query runtime with the specified tracer.

Source

pub fn query<Q: Query>(&self, query: Q) -> Result<Arc<Q::Output>, QueryError>

Execute a query synchronously.

Returns the cached result if valid, otherwise executes the query.

§Errors

QueryError::Suspend - Query is waiting for async loading
QueryError::Cycle - Dependency cycle detected

Source

pub fn invalidate<Q: Query>(&self, query: &Q)

Invalidate a query, forcing recomputation on next access.

This also invalidates any queries that depend on this one.

Source

pub fn remove_query<Q: Query>(&self, query: &Q)

Remove a query from the cache entirely, freeing memory.

Use this for GC when a query is no longer needed. Unlike invalidate, this removes all traces of the query from storage. The query will be recomputed from scratch on next access.

This also invalidates any queries that depend on this one.

Source

pub fn clear_cache(&self)

Clear all cached values by removing all nodes from whale.

Note: This is a relatively expensive operation as it iterates through all keys.

Source

pub fn poll<Q: Query>( &self, query: Q, ) -> Result<Polled<Result<Arc<Q::Output>, Arc<Error>>>, QueryError>

Poll a query, returning both the result and its change revision.

This is useful for implementing subscription patterns where you need to detect changes efficiently. Compare the returned revision with a previously stored value to determine if the query result has changed.

The returned Polled contains a Result<Arc<Q::Output>, Arc<anyhow::Error>> as its value, allowing you to track revision changes for both success and user error cases.

§Example

use query_flow::{asset_key, query, Db, DurabilityLevel, QueryError, QueryRuntime};

#[asset_key(asset = i32)]
struct Input(&'static str);

#[query]
fn doubled(db: &impl Db) -> Result<i32, QueryError> {
    Ok(*db.asset(Input("x"))? * 2)
}

let runtime = QueryRuntime::new();
runtime.resolve_asset(Input("x"), 21, DurabilityLevel::Volatile);

let result = runtime.poll(Doubled::new()).unwrap();
assert_eq!(**result.value.as_ref().unwrap(), 42);
let last_revision = result.revision;

// Polling again without any change leaves the revision untouched,
// so a subscriber knows there is nothing to send.
let again = runtime.poll(Doubled::new()).unwrap();
assert_eq!(again.revision, last_revision);

// Changing the input bumps the revision.
runtime.resolve_asset(Input("x"), 50, DurabilityLevel::Volatile);
let changed = runtime.poll(Doubled::new()).unwrap();
assert!(changed.revision > last_revision);
assert_eq!(**changed.value.as_ref().unwrap(), 100);

§Errors

Returns Err only for system errors (Suspend, Cycle, etc.). User errors are returned as Ok(Polled { value: Err(error), ... }).

Source

pub fn changed_at<Q: Query>(&self, query: &Q) -> Option<RevisionCounter>

Get the change revision of a query without executing it.

Returns None if the query has never been executed.

This is useful for checking if a query has changed since the last poll without the cost of executing the query.

§Example

use query_flow::{query, Db, QueryError, QueryRuntime, RevisionCounter};

#[query]
fn my_query(db: &impl Db, key: i32) -> Result<i32, QueryError> {
    let _ = db;
    Ok(key * 2)
}

let runtime = QueryRuntime::new();
let key = 21;
let last_known_revision: RevisionCounter = 0;

// Never executed yet, so there is no revision to compare against.
assert!(runtime.changed_at(&MyQuery::new(key)).is_none());

runtime.query(MyQuery::new(key)).unwrap();

// Check if query has changed before deciding to poll
if let Some(rev) = runtime.changed_at(&MyQuery::new(key)) {
    if rev > last_known_revision {
        let result = runtime.query(MyQuery::new(key)).unwrap();
        assert_eq!(*result, 42);
    }
}

Source §

impl<T: Tracer> QueryRuntime<T>

Source

pub fn query_keys(&self) -> Vec<FullCacheKey>

Get all query keys currently in the cache.

This is useful for implementing custom garbage collection strategies. Use this in combination with Tracer::on_query_key to track access times and implement LRU, TTL, or other GC algorithms externally.

§Example

use query_flow::{query, Db, QueryError, QueryRuntime};

#[query]
fn leaf(db: &impl Db, x: i32) -> Result<i32, QueryError> {
    let _ = db;
    Ok(x)
}

#[query]
fn root(db: &impl Db, x: i32) -> Result<i32, QueryError> {
    Ok(*db.query(Leaf::new(x))? + 1)
}

let runtime = QueryRuntime::new();
runtime.query(Root::new(1)).unwrap();
runtime.query(Root::new(2)).unwrap();

// Both roots and their dependencies are cached. Note that the returned
// keys also include the internal per-type set sentinels used by
// `list_queries`, so the count is larger than the number of queries.
assert!(runtime.query_keys().len() >= 4);

// Collect the keys that haven't been accessed recently. A real GC would
// consult access times recorded through `Tracer::on_query_key`; here
// every `Leaf` stands in for the stale set.
let stale_keys: Vec<_> = runtime
    .query_keys()
    .into_iter()
    .filter(|key| key.downcast::<Leaf>().is_some())
    .collect();
assert_eq!(stale_keys.len(), 2);

// The sweep keeps both leaves: each one still has a root depending on it,
// and `remove_if_unused` never breaks a live dependent.
for key in &stale_keys {
    assert!(!runtime.remove_if_unused(key));
}

// The roots themselves have no dependents, so they are reclaimed.
assert!(runtime.remove_query_if_unused(&Root::new(1)));

Source

pub fn remove_query_if_unused<Q: Query>(&self, query: &Q) -> bool

Remove a query if it has no dependents.

Returns true if the query was removed, false if it has dependents or doesn’t exist. This is the safe way to remove queries during GC, as it won’t break queries that depend on this one.

§Example

use query_flow::{query, Db, QueryError, QueryRuntime};

#[query]
fn leaf(db: &impl Db, x: i32) -> Result<i32, QueryError> {
    let _ = db;
    Ok(x)
}

#[query]
fn root(db: &impl Db, x: i32) -> Result<i32, QueryError> {
    Ok(*db.query(Leaf::new(x))? + 1)
}

let runtime = QueryRuntime::new();
runtime.query(Root::new(1)).unwrap();

// `Leaf` has a dependent (`Root`), so it is kept.
assert!(!runtime.remove_query_if_unused(&Leaf::new(1)));

// `Root` has no dependents, so it is removed.
assert!(runtime.remove_query_if_unused(&Root::new(1)));

Source

pub fn remove(&self, key: &FullCacheKey) -> bool

Remove a query by its FullCacheKey.

This is the type-erased version of remove_query. Use this when you have a FullCacheKey from query_keys or Tracer::on_query_key.

Returns true if the query was removed, false if it doesn’t exist.

§Warning

This forcibly removes the query even if other queries depend on it. Dependent queries will be recomputed on next access. For safe GC, use remove_if_unused instead.

Source

pub fn remove_if_unused(&self, key: &FullCacheKey) -> bool

Remove a query by its FullCacheKey if it has no dependents.

This is the type-erased version of remove_query_if_unused. Use this when you have a FullCacheKey from query_keys or Tracer::on_query_key.

Returns true if the query was removed, false if it has dependents or doesn’t exist.

§Example

use std::collections::HashSet;

use query_flow::{query, Db, FullCacheKey, QueryError, QueryRuntime};

#[query]
fn my_query(db: &impl Db, x: i32) -> Result<i32, QueryError> {
    let _ = db;
    Ok(x * 2)
}

let runtime = QueryRuntime::new();
runtime.query(MyQuery::new(1)).unwrap();
runtime.query(MyQuery::new(2)).unwrap();

// Your GC tracker decides what has expired; here everything has.
let expired: HashSet<FullCacheKey> = runtime.query_keys().into_iter().collect();

// Implement LRU GC
for key in runtime.query_keys() {
    if expired.contains(&key) {
        runtime.remove_if_unused(&key);
    }
}

assert!(runtime.query_keys().is_empty());

Source §

impl<T: Tracer> QueryRuntime<T>

Source

pub fn register_asset_locator<K, L>(&self, locator: L)
where K: AssetKey, L: AssetLocator<K>,

Register an asset locator for a specific asset key type.

Only one locator can be registered per key type. Later registrations replace earlier ones.

§Example

use query_flow::{
    asset_key, AssetLocator, Db, DurabilityLevel, LocateResult, QueryError, QueryRuntime,
};

#[asset_key(asset = String)]
struct FilePath(String);

struct InMemoryLocator {
    prefix: String,
}

impl AssetLocator<FilePath> for InMemoryLocator {
    fn locate(&self, db: &impl Db, key: &FilePath) -> Result<LocateResult<String>, QueryError> {
        let _ = db;
        Ok(LocateResult::Ready {
            value: format!("{}{}", self.prefix, key.0),
            durability: DurabilityLevel::Static,
        })
    }
}

let runtime = QueryRuntime::new();
runtime.register_asset_locator(InMemoryLocator {
    prefix: "/assets/".to_string(),
});

Source

pub fn pending_assets(&self) -> Vec<PendingAsset>

Get an iterator over pending asset requests.

Returns assets that have been requested but not yet resolved. The user should fetch these externally and call resolve_asset().

§Example

use query_flow::{
    asset_key, asset_locator, query, Db, DurabilityLevel, LocateResult, QueryError,
    QueryRuntime,
};

#[asset_key(asset = String)]
struct FilePath(String);

#[asset_locator]
fn pending(_db: &impl Db, _key: &FilePath) -> Result<LocateResult<String>, QueryError> {
    Ok(LocateResult::Pending)
}

#[query]
fn read_file(db: &impl Db, path: FilePath) -> Result<usize, QueryError> {
    Ok(db.asset(path)?.len())
}

fn fetch_file(path: &FilePath) -> String {
    format!("contents of {}", path.0)
}

let runtime = QueryRuntime::new();
runtime.register_asset_locator(Pending);

// The query suspends, which registers a pending asset request.
assert!(runtime
    .query(ReadFile::new(FilePath("a.txt".into())))
    .is_err());

for pending in runtime.pending_assets() {
    if let Some(path) = pending.key::<FilePath>() {
        let content = fetch_file(path);
        runtime.resolve_asset(path.clone(), content, DurabilityLevel::Volatile);
    }
}

assert_eq!(
    *runtime
        .query(ReadFile::new(FilePath("a.txt".into())))
        .unwrap(),
    17
);

Source

pub fn pending_assets_of<K: AssetKey>(&self) -> Vec<K>

Get pending assets filtered by key type.

Source

pub fn has_pending_assets(&self) -> bool

Check if there are any pending assets.

Source

pub fn resolve_asset<K: AssetKey>( &self, key: K, value: K::Asset, durability: DurabilityLevel, )

Resolve an asset with its loaded value.

This marks the asset as ready and invalidates any queries that depend on it (if the value changed), triggering recomputation on next access.

This method is idempotent - resolving with the same value (via asset_eq) will not trigger downstream recomputation.

§Arguments

key - The asset key identifying this resource
value - The loaded asset value
durability - How frequently this asset is expected to change

§Example

use query_flow::{asset_key, query, Db, DurabilityLevel, QueryError, QueryRuntime};

#[asset_key(asset = String)]
struct FilePath(String);

#[query]
fn byte_len(db: &impl Db, path: FilePath) -> Result<usize, QueryError> {
    Ok(db.asset(path)?.len())
}

let runtime = QueryRuntime::new();
let path = "config.json".to_string();

// In a real program this would be `std::fs::read_to_string(&path)?`.
let content = "hello".to_string();
runtime.resolve_asset(FilePath(path.clone()), content, DurabilityLevel::Volatile);

assert_eq!(*runtime.query(ByteLen::new(FilePath(path))).unwrap(), 5);

Source

pub fn resolve_asset_error<K: AssetKey>( &self, key: K, error: impl Into<Error>, durability: DurabilityLevel, )

Resolve an asset with an error.

This marks the asset as errored and caches the error. Queries depending on this asset will receive Err(QueryError::UserError(...)).

Use this when async loading fails (e.g., network error, file not found, access denied).

§Arguments

key - The asset key identifying this resource
error - The error to cache (will be wrapped in Arc)
durability - How frequently this error state is expected to change

§Example

use std::io;

use query_flow::{asset_key, query, Db, DurabilityLevel, QueryError, QueryRuntime};

#[asset_key(asset = String)]
struct FilePath(String);

#[query]
fn byte_len(db: &impl Db, path: FilePath) -> Result<usize, QueryError> {
    Ok(db.asset(path)?.len())
}

fn fetch_file(path: &str) -> io::Result<String> {
    Err(io::Error::new(io::ErrorKind::NotFound, path.to_string()))
}

let runtime = QueryRuntime::new();
let path = "missing.json".to_string();

match fetch_file(&path) {
    Ok(content) => runtime.resolve_asset(FilePath(path.clone()), content, DurabilityLevel::Volatile),
    Err(e) => runtime.resolve_asset_error(FilePath(path.clone()), e, DurabilityLevel::Volatile),
}

// Queries depending on the asset now observe the cached user error.
let err = runtime.query(ByteLen::new(FilePath(path))).unwrap_err();
assert!(err.is::<io::Error>());

Source

pub fn invalidate_asset<K: AssetKey>(&self, key: &K)

Invalidate an asset, forcing queries to re-request it.

The asset will be marked as loading and added to pending assets. Dependent queries will suspend until the asset is resolved again.

§Example

use query_flow::{asset_key, query, Db, DurabilityLevel, QueryError, QueryRuntime};

#[asset_key(asset = String)]
struct FilePath(String);

#[query]
fn byte_len(db: &impl Db, path: FilePath) -> Result<usize, QueryError> {
    Ok(db.asset(path)?.len())
}

let runtime = QueryRuntime::new();
let path = FilePath("config.json".into());
runtime.resolve_asset(path.clone(), "hello".into(), DurabilityLevel::Volatile);
assert_eq!(*runtime.query(ByteLen::new(path.clone())).unwrap(), 5);

// File was modified externally
runtime.invalidate_asset(&path);

// Queries depending on this asset will now suspend
let err = runtime.query(ByteLen::new(path.clone())).unwrap_err();
assert!(matches!(err, QueryError::Suspend { .. }));

// User should fetch the new value and call resolve_asset
runtime.resolve_asset(path.clone(), "hello world".into(), DurabilityLevel::Volatile);
assert_eq!(*runtime.query(ByteLen::new(path)).unwrap(), 11);

Source

pub fn remove_asset<K: AssetKey>(&self, key: &K)

Remove an asset from the cache entirely.

Unlike invalidate_asset, this removes all traces of the asset. Dependent queries will go through the locator again on next access.

Source

pub fn get_asset<K: AssetKey>( &self, key: K, ) -> Result<AssetLoadingState<K>, QueryError>

Get an asset by key without tracking dependencies.

Unlike QueryContext::asset(), this method does NOT register the caller as a dependent of the asset. Use this for direct asset access outside of query execution.

§Returns

Ok(AssetLoadingState::ready(...)) - Asset is loaded and ready
Ok(AssetLoadingState::loading(...)) - Asset is still loading (added to pending)
Err(QueryError::UserError) - Asset was not found or locator returned an error