Dio & Lens — Caching and Events

Chapter 4 gave you Vista — a universal handle that works with any backend. But every call still hits the backend. That was easy to forgive with a local SQLite file; this chapter picks a data source where it hurts: a real cloud API, hundreds of milliseconds away, read-only, and unable to sort or search. We’ll build a small CLI around it — an inventory of a public S3 bucket, with a cache.

Diorama (vantage-diorama) is the layer that makes this pleasant. It sits between your Vista and whatever consumes it, and does three things:

1. Transparent caching. Keep a local copy of your data. Reads come from cache, not from the master — a listing that costs seconds over the network costs microseconds after the first fetch.
2. Capability injection. A Vista backed by S3 can’t sort, search, or paginate a listing. Diorama caches the dataset locally and answers those queries from cache — the consumer sees a richer handle than the backend actually offers.
3. Custom write routing. Writes don’t have to go to the master — useful when the master can’t take them at all. You decide what a write means: a queue, a different store, an API call.

Table, Vista, Dio

This is the third handle to the same records — and like the query-vs-table comparison back in chapter 2, each one trades something away for something new:

Caching

Diorama caches at two levels. Page segments hold windows of an ordered query result — they’re what lets a viewport scroll a huge listing without re-asking the master, and they power the loading strategies you’ll meet in chapter 7. Beneath them sits the key/value record store — one entry per record id, plain and dumb. We keep it simple for now: everything in this chapter runs on the record store alone.

A Dio owns exactly the two blocks above: a master Vista — the source of truth, however far away — and a cache, the local copy. What it deliberately does not decide is policy: when the cache fills, when it goes stale, what a write means. There are too many valid answers for one default — seed once and keep forever, re-fetch on a timer, reconcile from a push stream, write through to the master, write somewhere else entirely. That’s what the Lens is for: it describes how the cache is used. The simplest possible Lens pumps the master into the cache once, when the Dio loads:

#![allow(unused)]
fn main() {
let lens = Arc::new(
    Lens::new()
        .cache_at("cache.redb")
        .on_start(|dio| {
            let dio = dio.clone();
            async move {
                // Pump: read everything from the master, write it to the cache.
                let rows = dio.master().list_values().await?;
                dio.cache().insert_values(rows).await
            }
        })
        .build()?,
);

let dio = lens.make_dio(vista).await?;   // runs on_start, returns a warm Dio
}

With the Dio in place, there are two ways to consume it. The first is the facade Vista — dio.vista() — for proactive querying: you ask, it answers, exactly like the backend’s own Vista from chapter 4. Same interface, same records — except the answers come from the cache, and the capability set can be wider than the master’s, because the Lens decides what to add: a read-only master gains writes when the Lens routes them somewhere, and counting is always on the menu because the cache can count. The second way in is the Scenery — a standing, reactive view that keeps itself current and tells you when it changed. UIs bind to sceneries; chapter 7 lives there.

The facade Vista

Hand dio.vista() to chapter 4’s print_vista and it just works — it never learns a cache is underneath. What makes the facade interesting is its capability set. Chapter 4’s contract was honest but rigid: whatever the backend can’t do, your application can’t have. The facade solves that — it carries the master’s capabilities, and the Lens can manipulate and extend them. Our read-only S3 listing gains can_insert the moment an on_write callback gives writes somewhere to go. The honesty contract still holds; the facade just advertises what the pipeline can do, not what the backend alone can.

The project: a weather-station inventory

NOAA publishes its daily climate archive — GHCN, one CSV per weather station — as a public S3 bucket in the AWS Open Data registry. It’s a perfect Diorama subject: listing it is slow, every listing request is paid again on every run, and the API can’t sort or search. We’ll grow one small tool across three chapters:

• This chapter — a CLI that lists the station files from a persistent local cache.
• Chapter 6 — augmentation: each file gains columns computed from its contents (how many readings, how recent).
• Chapter 7 — a live terminal UI that scrolls the whole archive and fetches per-file data for exactly the rows on screen.

Set up the project:

cargo new learn-4 && cd learn-4
cargo add serde --features derive
cargo add tokio --features full
cargo add vantage-aws vantage-diorama vantage-vista

vantage-aws is the S3/DynamoDB/IAM driver — it signs (or deliberately doesn’t sign) requests itself, so there’s no AWS SDK in the tree. vantage-diorama is what this chapter is about. The code splits into two files: files.rs holds the table definition, main.rs uses it.

files.rs — the listing as a table

One row per file in the bucket. This is chapter 2’s pattern — an entity, and a table() constructor that describes the source — pointed at a cloud API instead of a database:

#![allow(unused)]
fn main() {
use serde::{Deserialize, Serialize};
use vantage_aws::prelude::*;

/// One file in the bucket. Field names match S3's wire XML
/// (`<Contents><Key/><Size/></Contents>`).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct File {
    #[serde(rename = "Key")]
    pub filename: String,
    #[serde(rename = "Size")]
    pub size: String,
}
}

The serde renames map S3’s wire names onto the fields we actually want to write in Rust. Size is a String because that’s what the XML carries — no silent coercion.

#![allow(unused)]
fn main() {
impl File {
    /// `ListObjectsV2` narrowed to one bucket and prefix. S3 sends at
    /// most `max-keys` keys per response; the `@continuation-token`
    /// cursor tells the framework to keep requesting pages until the
    /// listing is complete.
    pub fn table(aws: AwsAccount, bucket: &str, prefix: &str) -> Table<AwsAccount, File> {
        Table::new(
            "restxml/Contents@continuation-token=NextContinuationToken:s3/GET /{Bucket}?list-type=2",
            aws,
        )
        .with_id_column("Key")
        .with_column_of::<String>("Size")
        .with_condition(eq("Bucket", bucket))
        .with_condition(eq("prefix", prefix))
        .with_condition(eq("max-keys", 100))
    }
}
}

The table name is doing a lot of work here — for vantage-aws, it’s the wire protocol spelled out: restxml is the protocol S3 speaks; Contents is the response element holding the rows; s3/GET /{Bucket}?list-type=2 is the service and request. Conditions complete the request the same way chapter 2’s conditions completed a WHERE clause: Bucket fills the path placeholder, and anything else (prefix, max-keys) becomes a query parameter.

The @continuation-token=NextContinuationToken part is auto-pagination. S3 answers at most max-keys files per response, plus a continuation token when more exist; the cursor declaration tells the driver to keep re-issuing the request — token folded back in — until the listing is complete. One list() call, as many HTTP requests as it takes.

{protocol}/{array_key}[@cursor|@request=response]:{service}/{METHOD} {path}?{query}

A first listing

main.rs, shortest possible version — build the table, list it, print:

mod files;

use files::File;
use vantage_aws::prelude::*;

const BUCKET: &str = "noaa-ghcn-pds";
const PREFIX: &str = "csv/by_station/GM";

#[tokio::main]
async fn main() {
    if let Err(e) = run().await {
        e.report();
    }
}

async fn run() -> VantageResult<()> {
    let aws = AwsAccount::public("us-east-1");
    let files = File::table(aws, BUCKET, PREFIX);

    for (filename, file) in files.list().await? {
        println!("{:>10}  {filename}", file.size);
    }
    Ok(())
}

The GM prefix narrows the archive to Germany’s stations — 1122 files, small enough to look at, big enough to feel:

   4796352  csv/by_station/GM000001153.csv
   7057976  csv/by_station/GM000001474.csv
   2978238  csv/by_station/GM000002277.csv
   ...
1122 files in 2.3s

2.3 seconds: twelve HTTPS round-trips (1122 files at 100 per page), paid on every run, forever, because nothing remembers the answer. This is the itch the rest of the chapter scratches.

The master Vista, and what it can’t do

The Dio wants a Vista for its master, so we erase the typed table the same way chapter 4 did:

#![allow(unused)]
fn main() {
let master = aws
    .vista_factory()
    .from_table(File::table(aws.clone(), BUCKET, PREFIX))?;
}

Ask this Vista what it can do, and you’ll see why this backend needs a Diorama:

#![allow(unused)]
fn main() {
let caps = master.capabilities();
// can_count: true    can_fetch_next: true — and that's it.
// can_search: false   can_order: false   can_insert: false
}

can_fetch_next is worth pausing on — it’s the one capability this chapter leans on. fetch_next(token) is Vista’s cursor-style pagination from chapter 4: pass None for the first page, pass back the returned token for the next one. For S3 the driver defines the token as the last key of the previous page, because S3 lists keys in order and accepts any key as a starting point (start-after). That has a consequence the generic contract doesn’t promise: this cursor survives process restarts. Any key you already hold — say, the last key in a cache — resumes the listing right after it.

The Lens: pump pages, resume where you left off

Now the Lens. One callback — on start, synchronize the cache with the bucket:

#![allow(unused)]
fn main() {
let lens = Arc::new(
    Lens::new()
        .cache_at("cache.redb")
        .on_start(|dio| {
            let dio = dio.clone();
            async move { sync(&dio).await }
        })
        .build()
        .context("Failed to build lens")?,
);
let dio = lens.make_dio(master).await?;
}

• .cache_at("cache.redb") opens a redb file on disk. This is why the cache survives process restarts. An in-memory alternative, .cache_in_memory(), is there for when persistence isn’t wanted.
• .on_start(...) fires once, inside make_dio. Callbacks receive &Dio and clone it to hold across .await — a cheap Arc bump; every Lens callback follows this shape.
• make_dio does the rest in one call: opens a cache table named after the master, spawns the write worker, runs on_start, and hands back the Dio.

sync is where the cursor pays off. Instead of one big list_values(), it pulls the listing one page at a time and seeds the cursor from the cache:

#![allow(unused)]
fn main() {
/// Pump the master listing into the cache, one page per request. S3's
/// paging cursor is simply "the last key seen" — so the last key already
/// in the cache resumes the listing, and pages loaded by an earlier run
/// (even one that was interrupted) are never fetched again.
async fn sync(dio: &Dio) -> VantageResult<()> {
    let mut token: Option<CborValue> = dio
        .cache()
        .list_values()
        .await?
        .keys()
        .last()
        .cloned()
        .map(Into::into);
    loop {
        let start = Instant::now();
        let (page, next) = dio.master().fetch_next(token).await?;
        let count = page.len();
        dio.cache().insert_values(page.into_iter().collect()).await?;
        println!("fetched {count} files in {:?}", start.elapsed());
        if next.is_none() {
            return Ok(());
        }
        token = next;
    }
}
}

Read the first statement again: the initial token is the last filename already cached. On a cold cache that’s None — start from the top. On a warm cache it’s the end of what we have — S3 continues from there. Kill the process halfway through a sync and run it again: the pages you already paid for are never fetched twice. The cache isn’t just an answer store; it’s the resume point.

Reads come from the cache

The listing loop stops asking S3 and starts asking the Dio — through the facade Vista from the first half of the chapter. Notice who talks to what: sync, being Lens plumbing, addresses the Dio’s two sides directly (dio.master() to fetch, dio.cache() to store and resume); a consumer asks dio.vista() and never learns what’s underneath. The records come back as Record<CborValue> — the erased form from chapter 4 — so field access goes through the CborValueExt helpers the prelude brings in:

#![allow(unused)]
fn main() {
let start = Instant::now();
let listing = dio.vista().list_values().await?;
for (filename, file) in &listing {
    let size = file.get("Size").and_then(|v| v.as_str()).unwrap_or("");
    println!("{size:>10}  {filename}");
}
println!("{} files from cache in {:?}", listing.len(), start.elapsed());
}

First run — cold cache, the sync pump narrating each page:

fetched 100 files in 215ms
fetched 100 files in 217ms
...
fetched 22 files in 114ms
   4796352  csv/by_station/GM000001153.csv
   7057976  csv/by_station/GM000001474.csv
   ...
1122 files from cache in 18ms

Second run — same command, new process:

fetched 0 files in 370ms
1122 files from cache in 17ms

One round-trip — the resume request from the last cached key, confirming nothing new exists — and then the full listing in 17 milliseconds. New files that do appear under the prefix arrive on exactly that request, without re-fetching the thousand we already hold.

Invalidating

A resuming cache has one blind spot: it only ever looks past what it holds, so files deleted from the bucket linger locally. The honest fix is to start over — wipe and re-pump:

#![allow(unused)]
fn main() {
if std::env::args().any(|a| a == "--invalidate") {
    dio.cache().clear().await?;
    sync(&dio).await?;
}
}

$ cargo run -- --invalidate
fetched 0 files in 370ms
fetched 100 files in 215ms
...
1122 files from cache in 18ms

The CLI is complete: two seconds once, milliseconds forever after, resumable mid-sync, and an escape hatch back to the truth.

The event bus

One more thing make_dio set up, invisibly: every Dio carries a broadcast event bus. Anything that changes data announces it there as a DioEvent — DatasetChanged when the set of records was rewritten wholesale, RecordChanged / RecordInserted / RecordRemoved for row-level changes, Refreshing when a reconcile starts, WriteFailed when a queued write fails. Nothing in this chapter listens yet; chapter 7’s sceneries subscribe to this bus to know when to recompute, and chapter 6’s hydration uses it to report progress.

One rule is worth forming as a habit now: direct cache writes are silent. Our sync calls dio.cache().insert_values(...), which stores rows and announces nothing — harmless while nothing listens, invisible data the moment something does. When observers exist, either use the Dio’s row-level helpers that write and announce in one motion — patched(id, record), removed(id) — or follow a bulk cache write with notify_dataset_changed(), the “re-read everything” announcement. Chapter 7’s live view does exactly that.

Callback summary

What we covered