Clean Provider object trait

crates/strata/src/dataset.rs

@@ -9,6 +9,7 @@
 use anyhow::{Result, bail};
 use futures::stream::{BoxStream, StreamExt};
 use schema::Schema;
+use serde::Serialize;
 use serde_json::Value;
 
 use crate::page::Cursor;
@@ -17,7 +18,7 @@ use crate::record::Records;
 /// How a sink should apply a written dataset. Rides as metadata on the existing
 /// `put` verb (the reserved `disposition` query param) rather than a new verb, so
 /// every sink shares one write surface.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize)]
 pub enum Disposition {
     /// Insert every row (the historical behavior). Re-running adds duplicates.
     #[default]

crates/strata/src/flight/mod.rs

@@ -55,12 +55,12 @@ impl StrataFlight {
             .get(provider)
             .map_err(|e| Status::not_found(e.to_string()))?;
 
-        let schema = provider
-            .resolve_schema(path)
+        let endpoint = provider
+            .resolve(path)
             .await
             .map_err(|e| Status::internal(e.to_string()))?;
 
-        Ok(strata_schema_to_arrow_schema(&schema))
+        Ok(strata_schema_to_arrow_schema(&endpoint.response))
     }
 }
 
@@ -164,10 +164,10 @@ impl FlightService for StrataFlight {
                 .map_err(|e| Status::internal(e.to_string()))?;
             // Advertise the static schema per endpoint; endpoints whose schema
             // is dynamic or not record-shaped (e.g. a generic `Row`) are skipped.
-            for (path, schema) in provider.endpoint_schemas() {
-                let arrow_schema = strata_schema_to_arrow_schema(&schema);
+            for endpoint in provider.endpoints() {
+                let arrow_schema = strata_schema_to_arrow_schema(&endpoint.response);
 
-                let descriptor = FlightDescriptor::new_path(vec![name.clone(), path]);
+                let descriptor = FlightDescriptor::new_path(vec![name.clone(), endpoint.path]);
                 if let Ok(info) = FlightInfo::new().try_with_schema(&arrow_schema) {
                     infos.push(Ok(info.with_descriptor(descriptor)));
                 }

crates/strata/src/graphql/mod.rs

@@ -62,12 +62,13 @@ async fn build_schema(registry: &Arc<Registry>) -> Result<Schema> {
         for table in tables(registry, &mount).await {
             let path = format!("/tables/{table}");
             let provider = registry.get(&mount)?;
-            if !provider.queryable(&path) {
-                continue;
-            }
-            let Ok(row_schema) = provider.resolve_schema(&path).await else {
+            let Ok(endpoint) = provider.resolve(&path).await else {
                 continue;
             };
+            if !endpoint.metadata.queryable {
+                continue;
+            }
+            let row_schema = endpoint.response;
             let type_name = format!("{mount}_{table}");
             objects.push(row_object(&type_name, &row_schema));
             query = query.field(table_field(

crates/strata/src/main.rs

@@ -168,8 +168,8 @@ async fn run() -> Result<()> {
         Command::List {
             provider: Some(name),
         } => {
-            for route in registry.get(&name)?.routes() {
-                println!("{route}");
+            for endpoint in registry.get(&name)?.endpoints() {
+                println!("{}", endpoint.path);
             }
         }
         Command::Schema { provider: None } => {
@@ -248,14 +248,14 @@ async fn run() -> Result<()> {
                     // unreachable at startup shouldn't stop `serve`.
                     match registry
                         .get(&pipe.source.mount)?
-                        .resolve_schema(&pipe.source.path)
+                        .resolve(&pipe.source.path)
                         .await
                     {
-                        Ok(schema) => {
+                        Ok(endpoint) => {
                             for ep in [&pipe.source, &pipe.destination] {
                                 let ep =
                                     strata_types::Endpoint::new(&ep.mount, strip_query(&ep.path));
-                                db.upsert_source_schema(&ep, &schema).await?;
+                                db.upsert_source_schema(&ep, &endpoint.response).await?;
                             }
                         }
                         Err(e) => tracing::warn!(

crates/strata/src/page.rs

@@ -48,7 +48,7 @@ impl<T> Page<T> {
 /// `list` route (`.strategy(…)`); the router only stores the signal — the sync
 /// system alongside `pipe` is what interprets it. The two differ in where a run
 /// starts, when it stops, and what a re-sync does.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
 pub enum ListStrategy {
     /// Finite backfill by offset/limit. `next` reaches `None` at the tail, so a run
     /// starts from the beginning and drains to completion. A re-sync re-scans from

crates/strata/src/pipe/mod.rs

@@ -19,7 +19,8 @@ use sync::Progress;
 pub async fn run_pass<S: PipeStore>(registry: &Registry, store: &S, pipe: &mut Pipe) -> Result<()> {
     let src = registry.get(&pipe.source.mount)?;
     let dst = registry.get(&pipe.destination.mount)?;
-    let strategy = src.strategy(&pipe.source.path).ok_or_else(|| {
+    let meta = src.resolve(&pipe.source.path).await?.metadata;
+    let strategy = meta.strategy.ok_or_else(|| {
         anyhow!(
             "source `{}{}` is not a list endpoint (no sync strategy)",
             pipe.source.mount,
@@ -36,7 +37,7 @@ pub async fn run_pass<S: PipeStore>(registry: &Registry, store: &S, pipe: &mut P
     // The source declares how a sink should apply its rows (merge if it re-emits
     // updates, else append). The sink still skips existing keys on append.
     let dst_path = WriteRequest::new(&pipe.destination.path)
-        .with_disposition(src.disposition(&pipe.source.path))
+        .with_disposition(meta.disposition)
         .path();
     let mut chunks = stream.chunks;
 

crates/strata/src/provider.rs

@@ -9,13 +9,11 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use anyhow::{Result, anyhow, bail};
-use schema::Schema;
 use serde_json::{Value, json};
 
 use crate::catalog::Catalog;
 use crate::config::ProviderConfig;
-use crate::dataset::{DataStream, Disposition};
-use crate::page::ListStrategy;
+use crate::dataset::DataStream;
 use crate::router::{Body, BoxFuture, EndpointInfo, Method, Response, Router};
 
 /// Implemented by each concrete provider. Knows its state type and how to wire
@@ -41,24 +39,11 @@ pub trait ProviderObject: Send + Sync {
     /// Wire the mount-scoped [`Catalog`] into this provider's router, so its
     /// handlers can read their own persisted annotations.
     fn set_catalog(&mut self, catalog: Catalog);
-    fn routes(&self) -> Vec<String>;
-    /// Machine-readable description of every endpoint (path, params, response
-    /// schema), for introspection.
+    /// Every endpoint, statically described (dynamic resolvers not run).
     fn endpoints(&self) -> Vec<EndpointInfo>;
-    /// Static response `DataType` of every endpoint, paired with its pattern.
-    fn endpoint_schemas(&self) -> Vec<(String, Schema)>;
-    /// Response `DataType` of the endpoint matching `path`, resolved (running a
-    /// dynamic schema resolver if the endpoint has one).
-    fn resolve_schema<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result<Schema>>;
-    /// The declared [`ListStrategy`] of the `list` route matching `path`, for the
-    /// external sync layer to drive its walk. `None` if no list route matches.
-    fn strategy(&self, path: &str) -> Option<ListStrategy>;
-    /// The declared write [`Disposition`] of the `list` route matching `path` —
-    /// whether a pipe should merge (source re-emits updates) or append. Defaults to
-    /// `Append`.
-    fn disposition(&self, path: &str) -> Disposition;
-    // TODO: Make this items just metadata map on the route
-    fn queryable(&self, path: &str) -> bool;
+    /// The read endpoint matching a concrete `path`, with its response schema
+    /// resolved (running a dynamic resolver if the route has one).
+    fn resolve<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result<EndpointInfo>>;
     /// Auto-pick the `List` read by path and return it as a [`DataStream`] — the
     /// data plane, for callers that consume the Arrow stream (pipe, CLI, Flight
     /// `do_get`). The router loops the provider's single-page handler internally.
@@ -92,39 +77,14 @@ where
         self.router.set_catalog(catalog);
     }
 
-    fn routes(&self) -> Vec<String> {
-        self.router.patterns()
-    }
-
     fn endpoints(&self) -> Vec<EndpointInfo> {
         self.router.endpoints()
     }
 
-    fn endpoint_schemas(&self) -> Vec<(String, Schema)> {
-        self.router.schemas()
-    }
-
-    fn resolve_schema<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result<Schema>> {
+    fn resolve<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result<EndpointInfo>> {
         let state = self.state.clone();
         let path = path.to_string();
-        Box::pin(async move {
-            self.router
-                .resolve_schema(state, &path)
-                .await
-                .ok_or_else(|| anyhow!("no endpoint matches `{path}`"))?
-        })
-    }
-
-    fn strategy(&self, path: &str) -> Option<ListStrategy> {
-        self.router.strategy(path)
-    }
-
-    fn disposition(&self, path: &str) -> Disposition {
-        self.router.disposition(path)
-    }
-
-    fn queryable(&self, path: &str) -> bool {
-        self.router.queryable(path)
+        Box::pin(async move { self.router.resolve(state, &path).await })
     }
 
     fn read<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result<DataStream>> {