diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index a78cc34140f..47a9ca2f757 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -1589,7 +1589,7 @@ impl TestHarness { maybe_claims: control_plane_api::MaybeControlClaims::with_verified(verified), original_uri: axum::http::Uri::from_static("/graphql"), pg_pool: self.pool.clone(), - refresh: app.snapshot.token(), + refresh: app.snapshot_watch.token(), retry_after: tokens::DateTime::UNIX_EPOCH, started: tokens::now(), locale: control_plane_api::Locale::EnUS, diff --git a/crates/control-plane-api/src/envelope.rs b/crates/control-plane-api/src/envelope.rs index 3ca7d720771..ef6b28cc57f 100644 --- a/crates/control-plane-api/src/envelope.rs +++ b/crates/control-plane-api/src/envelope.rs @@ -272,7 +272,7 @@ impl axum::extract::FromRequestParts> for Envelope { Ok(Envelope { maybe_claims, retry_after: retry_after.unwrap_or(tokens::DateTime::UNIX_EPOCH), - refresh: state.snapshot.token(), + refresh: state.snapshot_watch.token(), started: started.unwrap_or_else(|| tokens::now()), pg_pool: state.pg_pool.clone(), original_uri, diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs index 856766a693e..ca16775c095 100644 --- a/crates/control-plane-api/src/lib.rs +++ b/crates/control-plane-api/src/lib.rs @@ -53,7 +53,7 @@ pub use envelope::{Envelope, Locale, MaybeControlClaims}; pub(crate) use server::evaluate_names_authorization; pub use server::{ ApiError, App, AuthZRetry, build_router, - snapshot::{self, Snapshot}, + snapshot::{self, Authorization, Snapshot}, }; // Re-export the GraphQL schema SDL function for flow-client build script diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index e3eaef0fe71..406393688d2 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -39,7 +39,7 @@ pub struct App { pub control_plane_jwt_encode_key: tokens::jwt::EncodingKey, pub pg_pool: sqlx::PgPool, pub publisher: crate::publications::Publisher, - pub snapshot: Arc>, + pub snapshot_watch: Arc>, /// Signing secret for verifying inbound Stripe webhook deliveries. `None` /// when unconfigured, in which case the webhook endpoint fails closed rather /// than trusting any request. See `server::public::stripe_webhooks`. @@ -53,7 +53,7 @@ impl App { jwt_secret: &[u8], pg_pool: sqlx::PgPool, publisher: crate::publications::Publisher, - snapshot: Arc>, + snapshot_watch: Arc>, stripe_webhook_secret: Option, ) -> Self { Self { @@ -63,7 +63,7 @@ impl App { control_plane_jwt_encode_key: tokens::jwt::EncodingKey::from_secret(jwt_secret), pg_pool, publisher, - snapshot, + snapshot_watch, stripe_webhook_secret, } } diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs index 9c81400bfc0..2281a60ae3e 100644 --- a/crates/control-plane-api/src/server/snapshot.rs +++ b/crates/control-plane-api/src/server/snapshot.rs @@ -73,6 +73,37 @@ pub struct SnapshotTask { pub data_plane_id: models::Id, } +/// Outcome of an authorization check evaluated against a Snapshot, +/// classified by `Snapshot::resolve_authorization`. +#[must_use] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Authorization { + /// The required grant exists in the Snapshot. + Authorized, + /// The grant is absent and the Snapshot is authoritative for the + /// operation's anchor: the denial is final. + Denied, + /// The grant is absent but the Snapshot predates the anchor: a grant + /// committed before the anchor may not be reflected yet, so the denial is + /// provisional and the operation should retry under a fresher Snapshot. + Stale, +} + +impl Authorization { + /// Collapse to "is authorized?", surfacing a provisional denial as the + /// retryable `AuthorizationSnapshotStale` error which callers + /// (see `validation::is_authz_snapshot_stale`) convert into a retry. + pub fn ok_or_stale(self, catalog_name: &str) -> Result { + match self { + Authorization::Authorized => Ok(true), + Authorization::Denied => Ok(false), + Authorization::Stale => Err(validation::Error::AuthorizationSnapshotStale { + catalog_name: catalog_name.to_string(), + }), + } + } +} + // SnapshotMigration is the state of an underway data-plane migration. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub struct SnapshotMigration { @@ -181,6 +212,68 @@ impl Snapshot { self.taken > (started + Self::TEMPORAL_SKEW) } + /// Classify an already-evaluated authorization check against this + /// Snapshot's freshness: the single three-way policy — authorized / + /// authoritative denial / provisional denial — applied at every snapshot + /// authorization enforcement point. + /// + /// A denial is `Denied` only when this Snapshot was taken after `anchor`, + /// the instant the asking operation started: any grant committed before + /// the anchor is then necessarily reflected. Otherwise it is `Stale` — + /// possibly just unobserved. `None` means the caller has no instant to + /// anchor a staleness claim on, so denials are final. + pub fn resolve_authorization( + &self, + authorized: bool, + anchor: Option, + ) -> Authorization { + if authorized { + Authorization::Authorized + } else if anchor.is_none_or(|anchor| self.taken_after(anchor)) { + Authorization::Denied + } else { + Authorization::Stale + } + } + + /// Evaluate whether `user_id` holds `capability` to `name` under this + /// Snapshot's grants, classified against `anchor` freshness + /// (see `resolve_authorization`). + pub fn user_authorization( + &self, + user_id: uuid::Uuid, + name: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::UserGrant::is_authorized( + &self.role_grants, + &self.user_grants, + user_id, + name, + capability, + ), + anchor, + ) + } + + /// Evaluate whether `subject` (a catalog spec acting as a role) holds + /// `capability` to `object` under this Snapshot's role grants, classified + /// against `anchor` freshness (see `resolve_authorization`). + pub fn role_authorization( + &self, + subject: &str, + object: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::RoleGrant::is_authorized(&self.role_grants, subject, object, capability), + anchor, + ) + } + // Retrieve all tasks whose names start with the given `prefix`. pub fn tasks_by_prefix<'s>( &'s self, @@ -341,9 +434,26 @@ impl Snapshot { }) } + /// Returns the "spec capabilities" of a spec named `catalog_name`: the role + /// grants whose `subject_role` is a prefix of the name — the capabilities the + /// spec holds by virtue of its own name/role. This is only to be used for error + /// reporting to improve error messages. + pub fn spec_capabilities(&self, catalog_name: &str) -> Vec { + self.role_grants + .iter() + .filter(|grant| catalog_name.starts_with(grant.subject_role.as_str())) + .cloned() + .collect() + } + // Minimal interval between Snapshot refreshes. // We will postpone a requested refresh prior to this interval. pub const MIN_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::seconds(20); + /// Re-poll cadence for a queued task which is deferring until its + /// Snapshot is authoritative (see `taken_after`). This equals + /// `MIN_REFRESH_INTERVAL` because that's the soonest a refresh can land: + /// waking sooner burns polls, waking later delays the task. + pub const STALE_RETRY_WAKE: chrono::TimeDelta = Self::MIN_REFRESH_INTERVAL; // Maximum interval between Snapshot refreshes. // We will refresh an older Snapshot in the background. pub const MAX_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::minutes(5); @@ -837,4 +947,187 @@ mod tests { chrono::DateTime::from_timestamp(300_000, 0).unwrap() ); } + + /// `taken_after` is the single definition of "this Snapshot is authoritative + /// for that instant", and every authorization-staleness decision routes + /// through it. The `TEMPORAL_SKEW` allowance and the strictness of the + /// comparison are therefore load-bearing, so pin both. + #[test] + fn test_taken_after_allows_for_temporal_skew() { + let started = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let at = |offset: chrono::TimeDelta| Snapshot { + taken: started + offset, + ..Snapshot::empty() + }; + + assert!( + !at(chrono::TimeDelta::zero()).taken_after(started), + "a Snapshot taken at the same instant is not authoritative" + ); + assert!( + !at(-Snapshot::TEMPORAL_SKEW).taken_after(started), + "a Snapshot taken before the event is not authoritative" + ); + assert!( + !at(Snapshot::TEMPORAL_SKEW).taken_after(started), + "the skew allowance is exclusive: exactly TEMPORAL_SKEW later is still not authoritative" + ); + assert!( + at(Snapshot::TEMPORAL_SKEW + chrono::TimeDelta::milliseconds(1)).taken_after(started), + "one millisecond past the skew allowance is authoritative" + ); + } + + /// `resolve_authorization` is the shared three-way classifier behind every + /// snapshot authorization enforcement point. Pin its anchor semantics — + /// a denial is authoritative only under a Snapshot postdating the anchor, + /// and a `None` anchor makes denials final — and `ok_or_stale`'s collapse + /// into authorized / dropped / retryable. + #[test] + fn test_resolve_authorization() { + let anchor = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let stale = Snapshot { + taken: anchor, + ..Snapshot::empty() + }; + let fresh = Snapshot { + taken: anchor + Snapshot::TEMPORAL_SKEW * 2, + ..Snapshot::empty() + }; + + // A held grant is Authorized regardless of freshness. + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, Some(anchor)) + ); + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, None) + ); + + // A denial is authoritative only under a Snapshot postdating the anchor. + assert_eq!( + Authorization::Denied, + fresh.resolve_authorization(false, Some(anchor)) + ); + assert_eq!( + Authorization::Stale, + stale.resolve_authorization(false, Some(anchor)) + ); + + // Without an anchor there is no basis for a staleness claim. + assert_eq!( + Authorization::Denied, + stale.resolve_authorization(false, None) + ); + + assert!(matches!( + Authorization::Authorized.ok_or_stale("acmeCo/task"), + Ok(true) + )); + assert!(matches!( + Authorization::Denied.ok_or_stale("acmeCo/task"), + Ok(false) + )); + assert!(matches!( + Authorization::Stale.ok_or_stale("acmeCo/task"), + Err(validation::Error::AuthorizationSnapshotStale { catalog_name }) + if catalog_name == "acmeCo/task" + )); + } + + /// `spec_capabilities` replaced a SQL-computed `spec_capabilities` column and + /// now renders the "Available grants are:" list in publication authorization + /// errors. It answers "what may a spec named X do, by virtue of its own + /// name?", which is a prefix match on `subject_role` — not on `object_role`, + /// and not scoped to any user. + #[test] + fn test_spec_capabilities() { + let snapshot = Snapshot::build_fixture(None); + let subjects = |name: &str| { + snapshot + .spec_capabilities(name) + .into_iter() + .map(|g| { + ( + g.subject_role.to_string(), + g.object_role.to_string(), + g.capability, + ) + }) + .collect::>() + }; + + // A name under a granted prefix picks up every grant whose subject_role + // is a prefix of it — here both the tenant-wide grants and the more + // specific `bobCo/tires/` one. + insta::assert_debug_snapshot!(subjects("bobCo/tires/source-tread"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ( + "bobCo/tires/", + "acmeCo/shared/", + Read, + ), + ] + "#); + + // The narrower `bobCo/tires/` grant must not leak to a sibling prefix: + // subject matching is by prefix of the *name*, not by shared tenancy. + insta::assert_debug_snapshot!(subjects("bobCo/widgets/source-squash"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ] + "#); + + // `subject_role` is matched as a prefix of the name, so the role itself + // qualifies. + assert_eq!( + vec![( + "bobCo/tires/".to_string(), + "acmeCo/shared/".to_string(), + models::Capability::Read + )], + subjects("bobCo/tires/") + .into_iter() + .filter(|(s, _, _)| s == "bobCo/tires/") + .collect::>(), + ); + + // Grants are not matched by their object_role: `acmeCo/shared/` is + // reachable *from* `bobCo/tires/`, but a spec named `acmeCo/shared/x` + // holds only `acmeCo/`'s own grants. + insta::assert_debug_snapshot!(subjects("acmeCo/shared/thing"), @r#" + [ + ( + "acmeCo/", + "acmeCo/", + Write, + ), + ] + "#); + + assert!( + subjects("unknownCo/thing").is_empty(), + "a name under no granted prefix holds nothing" + ); + } } diff --git a/crates/data-plane-controller/src/shared/stack.rs b/crates/data-plane-controller/src/shared/stack.rs index a9642ebb41f..8d9454a7ecc 100644 --- a/crates/data-plane-controller/src/shared/stack.rs +++ b/crates/data-plane-controller/src/shared/stack.rs @@ -152,6 +152,13 @@ pub struct DataPlane { // IP at S3 changes from the NAT EIP to the private-subnet IP. #[serde(default, skip_serializing_if = "is_false")] pub s3_endpoint_on_private_subnet: bool, + // Extra tags applied to every AWS resource of the data-plane, through the + // Pulumi AWS provider's `default_tags`. Used for customer cost-allocation + // schemes which require a tag on resources we create and replace over + // time, such as the `map-migrated` tag of the AWS Migration Acceleration + // Program. AWS only; Azure and GCP deployments ignore it. + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub aws_resource_tags: std::collections::BTreeMap, pub deployments: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub connector_limits: Option, @@ -729,6 +736,40 @@ mod test { ); } + // We deserialize `data_planes.config` and re-serialize it into the Pulumi + // stack config, so `aws_resource_tags` reaches est-dry-dock only if it + // survives that round trip. An absent map must stay absent, so that + // data-planes which set no tags see no change to their stack config. + #[test] + fn aws_resource_tags_round_trip() { + let State { stack, .. } = serde_json::from_str(include_str!("state_fixture.json")).unwrap(); + + assert!(stack.config.model.aws_resource_tags.is_empty()); + assert_eq!( + serde_json::to_value(&stack.config.model) + .unwrap() + .get("aws_resource_tags"), + None, + ); + + let tagged = serde_json::to_value(DataPlane { + aws_resource_tags: [("map-migrated".to_string(), "migAQWO5XH8V0".to_string())].into(), + ..stack.config.model.clone() + }) + .unwrap(); + + assert_eq!( + tagged.get("aws_resource_tags").unwrap(), + &serde_json::json!({"map-migrated": "migAQWO5XH8V0"}), + ); + assert_eq!( + serde_json::from_value::(tagged) + .unwrap() + .aws_resource_tags, + [("map-migrated".to_string(), "migAQWO5XH8V0".to_string())].into(), + ); + } + // The wire shape est-dry-dock parses: `id` is a sibling of the flattened // link config, omitted when None, and round-trips losslessly. #[test] diff --git a/crates/models/src/shards.rs b/crates/models/src/shards.rs index a546fe6e348..ab594fa197b 100644 --- a/crates/models/src/shards.rs +++ b/crates/models/src/shards.rs @@ -32,7 +32,7 @@ pub struct ShardTemplate { /// may process documents before it must flush and commit. /// It may run for less time if there aren't additional ready documents for /// it to process. - /// If not set, the maximum duration defaults to five minutes for materializations, + /// If not set, the maximum duration defaults to twenty minutes for materializations, /// and one second for captures and derivations. /// EXPERIMENTAL: this field MAY be removed. #[serde( diff --git a/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap b/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap index de606792003..c2badf7ef95 100644 --- a/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap +++ b/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap @@ -1285,7 +1285,7 @@ expression: "&schema" }, "maxTxnDuration": { "title": "Maximum duration of task transactions.", - "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to five minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", + "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to twenty minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", "type": [ "string", "null" diff --git a/crates/validation/src/errors.rs b/crates/validation/src/errors.rs index 63fa9b2198f..187a5d17df6 100644 --- a/crates/validation/src/errors.rs +++ b/crates/validation/src/errors.rs @@ -299,6 +299,10 @@ pub enum Error { build_id: models::Id, larger_id: models::Id, }, + #[error( + "authorization for {catalog_name} was evaluated against a control-plane snapshot that is not authoritative for this operation; please retry the operation" + )] + AuthorizationSnapshotStale { catalog_name: String }, #[error( "This spec was updated while you were editing — please refresh and re-apply your changes.\nThis may have been an automated system update. (expected publication ID {expect_id}, actual {actual_id})" )] @@ -413,3 +417,15 @@ impl Error { errors.insert_row(scope.flatten(), anyhow::anyhow!(self)); } } + +/// Returns true if `err` is (or wraps) an [`Error::AuthorizationSnapshotStale`]. +/// This classifies a *retryable* authorization failure: the decision was made +/// against a control-plane snapshot that is not authoritative for the operation, +/// so it should be retried against a fresher snapshot rather than surfaced as a +/// terminal error. +pub fn is_authz_snapshot_stale(err: &anyhow::Error) -> bool { + matches!( + err.downcast_ref::(), + Some(Error::AuthorizationSnapshotStale { .. }) + ) +} diff --git a/crates/validation/src/lib.rs b/crates/validation/src/lib.rs index 1f9c0c31743..ea0b90f0f0c 100644 --- a/crates/validation/src/lib.rs +++ b/crates/validation/src/lib.rs @@ -17,7 +17,7 @@ mod schema; mod storage_mapping; mod test_step; -pub use errors::Error; +pub use errors::{Error, is_authz_snapshot_stale}; pub use noop::NoOpConnectors; /// Portion of the binding namespace reserved for runtime-internal bindings diff --git a/flow.schema.json b/flow.schema.json index 31b5fc6b359..4379b4f581b 100644 --- a/flow.schema.json +++ b/flow.schema.json @@ -1281,7 +1281,7 @@ }, "maxTxnDuration": { "title": "Maximum duration of task transactions.", - "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to five minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", + "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to twenty minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", "type": [ "string", "null" diff --git a/mise/tasks/build/rocksdb b/mise/tasks/build/rocksdb index 428c4832a9f..26063af33be 100755 --- a/mise/tasks/build/rocksdb +++ b/mise/tasks/build/rocksdb @@ -31,11 +31,21 @@ export ROCKSDB_DISABLE_LZ4=1 export ROCKSDB_DISABLE_ZLIB=1 export ROCKSDB_DISABLE_ZSTD=1 +# PORTABLE=0 means `-march=native`, so librocksdb.a inherited the builder's ISA: +# an AVX-512 runner once produced binaries that SIGILL everywhere else. haswell +# pins the AVX2 level we had been shipping. Avoid x86-64-v3 -- RocksDB falls back +# to a slower software checksum under it, since v3 lacks PCLMUL. aarch64 adds no +# -march under either setting, so 1 leaves it unchanged. +case "$(uname -m)" in + x86_64) PORTABLE_ARCH=haswell ;; + *) PORTABLE_ARCH=1 ;; +esac + make static_lib -j$(nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || echo 4) \ EXTRA_CXXFLAGS="-fPIC -frtti" \ EXTRA_CFLAGS="-fPIC" \ DEBUG_LEVEL=0 \ - PORTABLE=0 + PORTABLE="${PORTABLE_ARCH}" strip --strip-unneeded "${SOURCE_DIR}/librocksdb.a" diff --git a/site/README.md b/site/README.md index 3e39ea5429f..da53ce6d579 100644 --- a/site/README.md +++ b/site/README.md @@ -26,6 +26,15 @@ $ npm build This command generates static content into the `build` directory and can be served using any static contents hosting service. +**macOS note:** `npm run build` fails locally on macOS's default case-insensitive +filesystem with `The redirect plugin is not supposed to override existing files` +for the `hubspot-real-time` redirect in `docusaurus.config.js`. That redirect's +`from` path differs from the real `HubSpot-real-time` page only by case, so the +plugin sees a collision when writing the redirect stub. This does not affect +production: CI builds on Linux (case-sensitive), so the two paths are genuinely +distinct there. To build locally on macOS, temporarily comment out that redirect +entry, or build inside a case-sensitive volume/Docker. + ### Deployment Using SSH: diff --git a/site/docs/features/configuring-task-shards.md b/site/docs/features/configuring-task-shards.md index edd27f3342b..aa4e9384ecc 100644 --- a/site/docs/features/configuring-task-shards.md +++ b/site/docs/features/configuring-task-shards.md @@ -15,7 +15,7 @@ You do this by adding the `shards` configuration to the capture or materializati |---|---|---|---| | `/disable` | Disable | Disable processing of the task's shards. | Boolean | | `/logLevel` | Log level | Log levels may currently be \"error\", \"warn\", \"info\", \"debug\", or \"trace\". If not set, the effective log level is \"info\". | String | -| `/maxTxnDuration` | Maximum transaction duration | This duration upper-bounds the amount of time during which a transaction may process documents before it must initiate a commit. Note that it may take some additional time for the commit to complete after it is initiated. The shard may run for less time if there aren't additional ready documents for it to process. If not set, the maximum duration defaults to one second for captures and derivations, and 5 minutes for materializations. | String | +| `/maxTxnDuration` | Maximum transaction duration | This duration upper-bounds the amount of time during which a transaction may process documents before it must initiate a commit. Note that it may take some additional time for the commit to complete after it is initiated. The shard may run for less time if there aren't additional ready documents for it to process. If not set, the maximum duration defaults to one second for captures and derivations, and 20 minutes for materializations. | String | | `/minTxnDuration` | Minimum transaction duration | This duration lower-bounds the amount of time during which a transaction must process documents before it must flush and commit. It may run for more time if additional documents are available. The default value is zero seconds. | String | For more information about these controls and when you might need to use them, see: diff --git a/site/docs/getting-started/installation.md b/site/docs/getting-started/installation.md index 411d976b132..683ca60f88d 100644 --- a/site/docs/getting-started/installation.md +++ b/site/docs/getting-started/installation.md @@ -97,7 +97,7 @@ For a [GCS bucket](https://cloud.google.com/storage/docs/creating-buckets), upda * [`roles/storage.admin`](https://cloud.google.com/storage/docs/access-control/iam-roles) as the **role** You can configure this through the [Cloud Console](https://console.cloud.google.com/) or -the `gsutil` CLI. +the `gcloud storage` CLI. ### Amazon S3 diff --git a/site/docs/guides/customize-dataflows/materialization-sync-schedule.md b/site/docs/guides/customize-dataflows/materialization-sync-schedule.md index ae36c72eb0a..2816d72bcc8 100644 --- a/site/docs/guides/customize-dataflows/materialization-sync-schedule.md +++ b/site/docs/guides/customize-dataflows/materialization-sync-schedule.md @@ -18,8 +18,18 @@ reduce the costs incurred in the destination from the actions the connector takes to load data to it. ::: -Most warehouse materialization connectors (such as [Snowflake](/reference/Connectors/materialization-connectors/Snowflake), [Databricks](/reference/Connectors/materialization-connectors/databricks), and [BigQuery](/reference/Connectors/materialization-connectors/BigQuery)) support configuring a sync schedule. -Check the connector reference docs to determine if a specific connector supports sync schedules. +Sync schedules are supported by **warehouse** materialization connectors, such as [Snowflake](/reference/Connectors/materialization-connectors/Snowflake), [Databricks](/reference/Connectors/materialization-connectors/databricks), [BigQuery](/reference/Connectors/materialization-connectors/BigQuery), Amazon Redshift, MotherDuck, ClickHouse, and Azure Fabric Warehouse. + +In contrast, transactional database materializations (such as PostgreSQL, MySQL, and SQL Server) and streaming or API destinations (such as Elasticsearch, MongoDB, DynamoDB, and Pinecone) do _not_ use a sync schedule; they apply updates as they arrive. + +The connector reference docs will note whether a specific connector supports sync schedules. + +:::note +File and object-store connectors support a similar concept to sync schedules. +These connectors (such as the Amazon S3, Google Cloud Storage, and Azure Blob Storage file materializations) use an `uploadInterval` property or similar to manage data load frequency. + +They do not support other sync schedule functionality like fast sync times. +::: ## How transactions are used to sync data to a destination @@ -43,6 +53,15 @@ fully caught up - backfills always run as fast as possible. And while a transaction is delayed, Estuary will continue batching and combining new documents so that the next transaction contains all of the latest data. +The connector decides whether it is caught up or still backfilling by looking at +the sizes of its recent transactions. If any of the last several transactions +stored a large number of documents (on the order of a million), the +materialization is assumed to still be backfilling, and the sync schedule delay +is skipped so that it can catch up as fast as possible. Once several consecutive +transactions are all below that size, the materialization is treated as caught +up and the sync schedule delay applies. This behavior is the same for every +connector that supports a sync schedule or upload interval. + You can read about [how continuous materialization works](/concepts/materialization/#how-continuous-materialization-works) for more background information. diff --git a/site/docusaurus.config.js b/site/docusaurus.config.js index 61d85b9c638..183be88ec7d 100644 --- a/site/docusaurus.config.js +++ b/site/docusaurus.config.js @@ -46,7 +46,7 @@ const config = { tagline: 'Fast, configurable ETL your way', url: URL, baseUrl: BASE_URL, - onBrokenAnchors: 'warn', // TODO(johnny): Fix broken links and make this 'throw'. + onBrokenAnchors: 'throw', onBrokenLinks: 'throw', favicon: 'img/favicon-2.ico', organizationName: 'estuary',