mz_storage/source/
source_reader_pipeline.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Types related to the creation of dataflow raw sources.
11//!
12//! Raw sources are differential dataflow  collections of data directly produced by the
13//! upstream service. The main export of this module is [`create_raw_source`],
14//! which turns [`RawSourceCreationConfig`]s into the aforementioned streams.
15//!
16//! The full source, which is the _differential_ stream that represents the actual object
17//! created by a `CREATE SOURCE` statement, is created by composing
18//! [`create_raw_source`] with
19//! decoding, `SourceEnvelope` rendering, and more.
20//!
21
22// https://github.com/tokio-rs/prost/issues/237
23#![allow(missing_docs)]
24#![allow(clippy::needless_borrow)]
25
26use std::cell::RefCell;
27use std::collections::{BTreeMap, VecDeque};
28use std::convert::Infallible;
29use std::hash::{Hash, Hasher};
30use std::rc::Rc;
31use std::sync::Arc;
32use std::time::Duration;
33
34use differential_dataflow::lattice::Lattice;
35use differential_dataflow::{AsCollection, Collection, Hashable};
36use futures::stream::StreamExt;
37use mz_ore::cast::CastFrom;
38use mz_ore::collections::CollectionExt;
39use mz_ore::now::NowFn;
40use mz_persist_client::cache::PersistClientCache;
41use mz_repr::{Diff, GlobalId, RelationDesc, Row};
42use mz_storage_types::configuration::StorageConfiguration;
43use mz_storage_types::controller::CollectionMetadata;
44use mz_storage_types::dyncfgs;
45use mz_storage_types::errors::DataflowError;
46use mz_storage_types::sources::{SourceConnection, SourceExport, SourceTimestamp};
47use mz_timely_util::antichain::AntichainExt;
48use mz_timely_util::builder_async::{
49    Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
50};
51use mz_timely_util::capture::PusherCapture;
52use mz_timely_util::operator::ConcatenateFlatten;
53use mz_timely_util::reclock::reclock;
54use timely::PartialOrder;
55use timely::container::CapacityContainerBuilder;
56use timely::dataflow::channels::pact::Pipeline;
57use timely::dataflow::operators::capture::capture::Capture;
58use timely::dataflow::operators::capture::{Event, EventPusher};
59use timely::dataflow::operators::core::Map as _;
60use timely::dataflow::operators::generic::builder_rc::OperatorBuilder as OperatorBuilderRc;
61use timely::dataflow::operators::{Broadcast, CapabilitySet, Inspect, Leave};
62use timely::dataflow::scopes::Child;
63use timely::dataflow::{Scope, Stream};
64use timely::order::TotalOrder;
65use timely::progress::frontier::MutableAntichain;
66use timely::progress::{Antichain, Timestamp};
67use tokio::sync::{Semaphore, watch};
68use tokio_stream::wrappers::WatchStream;
69use tracing::trace;
70
71use crate::healthcheck::{HealthStatusMessage, HealthStatusUpdate};
72use crate::metrics::StorageMetrics;
73use crate::metrics::source::SourceMetrics;
74use crate::source::probe;
75use crate::source::reclock::ReclockOperator;
76use crate::source::types::{Probe, SourceMessage, SourceOutput, SourceRender, StackedCollection};
77use crate::statistics::SourceStatistics;
78
79/// Shared configuration information for all source types. This is used in the
80/// `create_raw_source` functions, which produce raw sources.
81#[derive(Clone)]
82pub struct RawSourceCreationConfig {
83    /// The name to attach to the underlying timely operator.
84    pub name: String,
85    /// The ID of this instantiation of this source.
86    pub id: GlobalId,
87    /// The details of the outputs from this ingestion.
88    pub source_exports: BTreeMap<GlobalId, SourceExport<CollectionMetadata>>,
89    /// The ID of the worker on which this operator is executing
90    pub worker_id: usize,
91    /// The total count of workers
92    pub worker_count: usize,
93    /// Granularity with which timestamps should be closed (and capabilities
94    /// downgraded).
95    pub timestamp_interval: Duration,
96    /// The function to return a now time.
97    pub now_fn: NowFn,
98    /// The metrics & registry that each source instantiates.
99    pub metrics: StorageMetrics,
100    /// The upper frontier this source should resume ingestion at
101    pub as_of: Antichain<mz_repr::Timestamp>,
102    /// For each source export, the upper frontier this source should resume ingestion at in the
103    /// system time domain.
104    pub resume_uppers: BTreeMap<GlobalId, Antichain<mz_repr::Timestamp>>,
105    /// For each source export, the upper frontier this source should resume ingestion at in the
106    /// source time domain.
107    ///
108    /// Since every source has a different timestamp type we carry the timestamps of this frontier
109    /// in an encoded `Vec<Row>` form which will get decoded once we reach the connection
110    /// specialized functions.
111    pub source_resume_uppers: BTreeMap<GlobalId, Vec<Row>>,
112    /// A handle to the persist client cache
113    pub persist_clients: Arc<PersistClientCache>,
114    /// Collection of `SourceStatistics` for source and exports to share updates.
115    pub statistics: BTreeMap<GlobalId, SourceStatistics>,
116    /// Enables reporting the remap operator's write frontier.
117    pub shared_remap_upper: Rc<RefCell<Antichain<mz_repr::Timestamp>>>,
118    /// Configuration parameters, possibly from LaunchDarkly
119    pub config: StorageConfiguration,
120    /// The ID of this source remap/progress collection.
121    pub remap_collection_id: GlobalId,
122    /// The storage metadata for the remap/progress collection
123    pub remap_metadata: CollectionMetadata,
124    // A semaphore that should be acquired by async operators in order to signal that upstream
125    // operators should slow down.
126    pub busy_signal: Arc<Semaphore>,
127}
128
129/// Reduced version of [`RawSourceCreationConfig`] that is used when rendering
130/// each export.
131#[derive(Clone)]
132pub struct SourceExportCreationConfig {
133    /// The ID of this instantiation of this source.
134    pub id: GlobalId,
135    /// The ID of the worker on which this operator is executing
136    pub worker_id: usize,
137    /// The metrics & registry that each source instantiates.
138    pub metrics: StorageMetrics,
139    /// Place to share statistics updates with storage state.
140    pub source_statistics: SourceStatistics,
141}
142
143impl RawSourceCreationConfig {
144    /// Returns the worker id responsible for handling the given partition.
145    pub fn responsible_worker<P: Hash>(&self, partition: P) -> usize {
146        let mut h = std::hash::DefaultHasher::default();
147        (self.id, partition).hash(&mut h);
148        let key = usize::cast_from(h.finish());
149        key % self.worker_count
150    }
151
152    /// Returns true if this worker is responsible for handling the given partition.
153    pub fn responsible_for<P: Hash>(&self, partition: P) -> bool {
154        self.responsible_worker(partition) == self.worker_id
155    }
156}
157
158/// Creates a source dataflow operator graph from a source connection. The type of SourceConnection
159/// determines the type of connection that _should_ be created.
160///
161/// This is also the place where _reclocking_
162/// (<https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/20210714_reclocking.md>)
163/// happens.
164///
165/// See the [`source` module docs](crate::source) for more details about how raw
166/// sources are used.
167///
168/// The `resume_stream` parameter will contain frontier updates whenever times are durably
169/// recorded which allows the ingestion to release upstream resources.
170pub fn create_raw_source<'g, G: Scope<Timestamp = ()>, C>(
171    scope: &mut Child<'g, G, mz_repr::Timestamp>,
172    storage_state: &crate::storage_state::StorageState,
173    committed_upper: &Stream<Child<'g, G, mz_repr::Timestamp>, ()>,
174    config: &RawSourceCreationConfig,
175    source_connection: C,
176    start_signal: impl std::future::Future<Output = ()> + 'static,
177) -> (
178    BTreeMap<
179        GlobalId,
180        Collection<
181            Child<'g, G, mz_repr::Timestamp>,
182            Result<SourceOutput<C::Time>, DataflowError>,
183            Diff,
184        >,
185    >,
186    Stream<G, HealthStatusMessage>,
187    Vec<PressOnDropButton>,
188)
189where
190    C: SourceConnection + SourceRender + Clone + 'static,
191{
192    let worker_id = config.worker_id;
193    let id = config.id;
194
195    let mut tokens = vec![];
196
197    let (ingested_upper_tx, ingested_upper_rx) =
198        watch::channel(MutableAntichain::new_bottom(C::Time::minimum()));
199    let (probed_upper_tx, probed_upper_rx) = watch::channel(None);
200
201    let source_metrics = Arc::new(config.metrics.get_source_metrics(id, worker_id));
202
203    let timestamp_desc = source_connection.timestamp_desc();
204
205    let (remap_collection, remap_token) = remap_operator(
206        scope,
207        storage_state,
208        config.clone(),
209        probed_upper_rx,
210        ingested_upper_rx,
211        timestamp_desc,
212    );
213    // Need to broadcast the remap changes to all workers.
214    let remap_collection = remap_collection.inner.broadcast().as_collection();
215    tokens.push(remap_token);
216
217    let committed_upper = reclock_committed_upper(
218        &remap_collection,
219        config.as_of.clone(),
220        committed_upper,
221        id,
222        Arc::clone(&source_metrics),
223    );
224
225    let mut reclocked_exports = BTreeMap::new();
226
227    let reclocked_exports2 = &mut reclocked_exports;
228    let (health, source_tokens) = scope.parent.scoped("SourceTimeDomain", move |scope| {
229        let (exports, source_upper, health_stream, source_tokens) = source_render_operator(
230            scope,
231            config,
232            source_connection,
233            probed_upper_tx,
234            committed_upper,
235            start_signal,
236        );
237
238        for (id, export) in exports {
239            let (reclock_pusher, reclocked) = reclock(&remap_collection, config.as_of.clone());
240            export
241                .inner
242                .map(move |(result, from_time, diff)| {
243                    let result = match result {
244                        Ok(msg) => Ok(SourceOutput {
245                            key: msg.key.clone(),
246                            value: msg.value.clone(),
247                            metadata: msg.metadata.clone(),
248                            from_time: from_time.clone(),
249                        }),
250                        Err(err) => Err(err.clone()),
251                    };
252                    (result, from_time.clone(), *diff)
253                })
254                .capture_into(PusherCapture(reclock_pusher));
255            reclocked_exports2.insert(id, reclocked);
256        }
257
258        source_upper.capture_into(FrontierCapture(ingested_upper_tx));
259
260        (health_stream.leave(), source_tokens)
261    });
262
263    tokens.extend(source_tokens);
264
265    (reclocked_exports, health, tokens)
266}
267
268pub struct FrontierCapture<T>(watch::Sender<MutableAntichain<T>>);
269
270impl<T: Timestamp> EventPusher<T, Vec<Infallible>> for FrontierCapture<T> {
271    fn push(&mut self, event: Event<T, Vec<Infallible>>) {
272        match event {
273            Event::Progress(changes) => self.0.send_modify(|frontier| {
274                frontier.update_iter(changes);
275            }),
276            Event::Messages(_, _) => unreachable!(),
277        }
278    }
279}
280
281/// Renders the source dataflow fragment from the given [SourceConnection]. This returns a
282/// collection timestamped with the source specific timestamp type. Also returns a second stream
283/// that can be used to learn about the `source_upper` that all the source reader instances know
284/// about. This second stream will be used by the `remap_operator` to mint new timestamp bindings
285/// into the remap shard.
286fn source_render_operator<G, C>(
287    scope: &mut G,
288    config: &RawSourceCreationConfig,
289    source_connection: C,
290    probed_upper_tx: watch::Sender<Option<Probe<C::Time>>>,
291    resume_uppers: impl futures::Stream<Item = Antichain<C::Time>> + 'static,
292    start_signal: impl std::future::Future<Output = ()> + 'static,
293) -> (
294    BTreeMap<GlobalId, StackedCollection<G, Result<SourceMessage, DataflowError>>>,
295    Stream<G, Infallible>,
296    Stream<G, HealthStatusMessage>,
297    Vec<PressOnDropButton>,
298)
299where
300    G: Scope<Timestamp = C::Time>,
301    C: SourceRender + 'static,
302{
303    let source_id = config.id;
304    let worker_id = config.worker_id;
305    let now_fn = config.now_fn.clone();
306    let timestamp_interval = config.timestamp_interval;
307
308    let resume_uppers = resume_uppers.inspect(move |upper| {
309        let upper = upper.pretty();
310        trace!(%upper, "timely-{worker_id} source({source_id}) received resume upper");
311    });
312
313    let (exports, progress, health, probes, tokens) =
314        source_connection.render(scope, config, resume_uppers, start_signal);
315
316    let mut export_collections = BTreeMap::new();
317
318    let source_metrics = config.metrics.get_source_metrics(config.id, worker_id);
319
320    // Compute the overall resume upper to report for the ingestion
321    let resume_upper = Antichain::from_iter(
322        config
323            .resume_uppers
324            .values()
325            .flat_map(|f| f.iter().cloned()),
326    );
327    source_metrics
328        .resume_upper
329        .set(mz_persist_client::metrics::encode_ts_metric(&resume_upper));
330
331    let mut health_streams = vec![];
332
333    for (id, export) in exports {
334        let name = format!("SourceGenericStats({})", id);
335        let mut builder = OperatorBuilderRc::new(name, scope.clone());
336
337        let (mut health_output, derived_health) =
338            builder.new_output::<CapacityContainerBuilder<_>>();
339        health_streams.push(derived_health);
340
341        let (mut output, new_export) = builder.new_output::<CapacityContainerBuilder<_>>();
342
343        let mut input = builder.new_input(&export.inner, Pipeline);
344        export_collections.insert(id, new_export.as_collection());
345
346        let bytes_read_counter = config.metrics.source_defs.bytes_read.clone();
347        let source_statistics = config
348            .statistics
349            .get(&id)
350            .expect("statistics initialized")
351            .clone();
352
353        builder.build(move |mut caps| {
354            let mut health_cap = Some(caps.remove(0));
355
356            move |frontiers| {
357                let mut last_status = None;
358                let mut health_output = health_output.activate();
359
360                if frontiers[0].is_empty() {
361                    health_cap = None;
362                    return;
363                }
364                let health_cap = health_cap.as_mut().unwrap();
365
366                while let Some((cap, data)) = input.next() {
367                    for (message, _, _) in data.iter() {
368                        let status = match &message {
369                            Ok(_) => HealthStatusUpdate::running(),
370                            // All errors coming into the data stream are definite.
371                            // Downstream consumers of this data will preserve this
372                            // status.
373                            Err(error) => HealthStatusUpdate::stalled(
374                                error.to_string(),
375                                Some(
376                                    "retracting the errored value may resume the source"
377                                        .to_string(),
378                                ),
379                            ),
380                        };
381
382                        let status = HealthStatusMessage {
383                            id: Some(id),
384                            namespace: C::STATUS_NAMESPACE.clone(),
385                            update: status,
386                        };
387                        if last_status.as_ref() != Some(&status) {
388                            last_status = Some(status.clone());
389                            health_output.session(&health_cap).give(status);
390                        }
391
392                        match message {
393                            Ok(message) => {
394                                source_statistics.inc_messages_received_by(1);
395                                let key_len = u64::cast_from(message.key.byte_len());
396                                let value_len = u64::cast_from(message.value.byte_len());
397                                bytes_read_counter.inc_by(key_len + value_len);
398                                source_statistics.inc_bytes_received_by(key_len + value_len);
399                            }
400                            Err(_) => {}
401                        }
402                    }
403                    let mut output = output.activate();
404                    output.session(&cap).give_container(data);
405                }
406            }
407        });
408    }
409
410    let probe_stream = match probes {
411        Some(stream) => stream,
412        None => synthesize_probes(source_id, &progress, timestamp_interval, now_fn),
413    };
414
415    // Broadcasting does more work than necessary, which would be to exchange the probes to the
416    // worker that will be the one minting the bindings but we'd have to thread this information
417    // through and couple the two functions enough that it's not worth the optimization (I think).
418    probe_stream.broadcast().inspect(move |probe| {
419        // We don't care if the receiver is gone
420        let _ = probed_upper_tx.send(Some(probe.clone()));
421    });
422
423    (
424        export_collections,
425        progress,
426        health.concatenate_flatten::<_, CapacityContainerBuilder<_>>(health_streams),
427        tokens,
428    )
429}
430
431/// Mints new contents for the remap shard based on summaries about the source
432/// upper it receives from the raw reader operators.
433///
434/// Only one worker will be active and write to the remap shard. All source
435/// upper summaries will be exchanged to it.
436fn remap_operator<G, FromTime>(
437    scope: &G,
438    storage_state: &crate::storage_state::StorageState,
439    config: RawSourceCreationConfig,
440    mut probed_upper: watch::Receiver<Option<Probe<FromTime>>>,
441    mut ingested_upper: watch::Receiver<MutableAntichain<FromTime>>,
442    remap_relation_desc: RelationDesc,
443) -> (Collection<G, FromTime, Diff>, PressOnDropButton)
444where
445    G: Scope<Timestamp = mz_repr::Timestamp>,
446    FromTime: SourceTimestamp,
447{
448    let RawSourceCreationConfig {
449        name,
450        id,
451        source_exports: _,
452        worker_id,
453        worker_count,
454        timestamp_interval,
455        remap_metadata,
456        as_of,
457        resume_uppers: _,
458        source_resume_uppers: _,
459        metrics: _,
460        now_fn,
461        persist_clients,
462        statistics: _,
463        shared_remap_upper,
464        config: _,
465        remap_collection_id,
466        busy_signal: _,
467    } = config;
468
469    let read_only_rx = storage_state.read_only_rx.clone();
470    let error_handler = storage_state.error_handler("remap_operator", id);
471
472    let chosen_worker = usize::cast_from(id.hashed() % u64::cast_from(worker_count));
473    let active_worker = chosen_worker == worker_id;
474
475    let operator_name = format!("remap({})", id);
476    let mut remap_op = AsyncOperatorBuilder::new(operator_name, scope.clone());
477    let (remap_output, remap_stream) = remap_op.new_output::<CapacityContainerBuilder<_>>();
478
479    let button = remap_op.build(move |capabilities| async move {
480        if !active_worker {
481            // This worker is not writing, so make sure it's "taken out" of the
482            // calculation by advancing to the empty frontier.
483            shared_remap_upper.borrow_mut().clear();
484            return;
485        }
486
487        let mut cap_set = CapabilitySet::from_elem(capabilities.into_element());
488
489        let remap_handle = crate::source::reclock::compat::PersistHandle::<FromTime, _>::new(
490            Arc::clone(&persist_clients),
491            read_only_rx,
492            remap_metadata.clone(),
493            as_of.clone(),
494            shared_remap_upper,
495            id,
496            "remap",
497            worker_id,
498            worker_count,
499            remap_relation_desc,
500            remap_collection_id,
501        )
502        .await;
503
504        let remap_handle = match remap_handle {
505            Ok(handle) => handle,
506            Err(e) => {
507                error_handler
508                    .report_and_stop(
509                        e.context(format!("Failed to create remap handle for source {name}")),
510                    )
511                    .await
512            }
513        };
514
515        let (mut timestamper, mut initial_batch) = ReclockOperator::new(remap_handle).await;
516
517        // Emit initial snapshot of the remap_shard, bootstrapping
518        // downstream reclock operators.
519        trace!(
520            "timely-{worker_id} remap({id}) emitting remap snapshot: trace_updates={:?}",
521            &initial_batch.updates
522        );
523
524        let cap = cap_set.delayed(cap_set.first().unwrap());
525        remap_output.give_container(&cap, &mut initial_batch.updates);
526        drop(cap);
527        cap_set.downgrade(initial_batch.upper);
528
529        let mut ticker = tokio::time::interval(timestamp_interval);
530        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
531
532        let mut prev_probe_ts: Option<mz_repr::Timestamp> = None;
533        let timestamp_interval_ms: u64 = timestamp_interval
534            .as_millis()
535            .try_into()
536            .expect("huge duration");
537
538        while !cap_set.is_empty() {
539            // Check the reclocking strategy in every iteration, to make it possible to change it
540            // without restarting the source pipeline.
541            let reclock_to_latest =
542                dyncfgs::STORAGE_RECLOCK_TO_LATEST.get(&config.config.config_set());
543
544            // If we are reclocking to the latest offset then we only mint bindings after a
545            // successful probe. Otherwise we fall back to the earlier behavior where we just
546            // record the ingested frontier.
547            let mut new_probe = None;
548            if reclock_to_latest {
549                new_probe = probed_upper
550                    .wait_for(|new_probe| match (prev_probe_ts, new_probe) {
551                        (None, Some(_)) => true,
552                        (Some(prev_ts), Some(new)) => prev_ts < new.probe_ts,
553                        _ => false,
554                    })
555                    .await
556                    .map(|probe| (*probe).clone())
557                    .unwrap_or_else(|_| {
558                        Some(Probe {
559                            probe_ts: now_fn().into(),
560                            upstream_frontier: Antichain::new(),
561                        })
562                    });
563            } else {
564                while prev_probe_ts >= new_probe.as_ref().map(|p| p.probe_ts) {
565                    ticker.tick().await;
566                    // We only proceed if the source upper frontier is not the minimum frontier. This
567                    // makes it so the first binding corresponds to the snapshot of the source, and
568                    // because the first binding always maps to the minimum *target* frontier we
569                    // guarantee that the source will never appear empty.
570                    let upstream_frontier = ingested_upper
571                        .wait_for(|f| *f.frontier() != [FromTime::minimum()])
572                        .await
573                        .unwrap()
574                        .frontier()
575                        .to_owned();
576
577                    let now = (now_fn)();
578                    let mut probe_ts = now - now % timestamp_interval_ms;
579                    if (now % timestamp_interval_ms) != 0 {
580                        probe_ts += timestamp_interval_ms;
581                    }
582                    new_probe = Some(Probe {
583                        probe_ts: probe_ts.into(),
584                        upstream_frontier,
585                    });
586                }
587            };
588
589            let probe = new_probe.expect("known to be Some");
590            prev_probe_ts = Some(probe.probe_ts);
591
592            let binding_ts = probe.probe_ts;
593            let cur_source_upper = probe.upstream_frontier;
594
595            let new_into_upper = Antichain::from_elem(binding_ts.step_forward());
596
597            let mut remap_trace_batch = timestamper
598                .mint(binding_ts, new_into_upper, cur_source_upper.borrow())
599                .await;
600
601            trace!(
602                "timely-{worker_id} remap({id}) minted new bindings: \
603                updates={:?} \
604                source_upper={} \
605                trace_upper={}",
606                &remap_trace_batch.updates,
607                cur_source_upper.pretty(),
608                remap_trace_batch.upper.pretty()
609            );
610
611            let cap = cap_set.delayed(cap_set.first().unwrap());
612            remap_output.give_container(&cap, &mut remap_trace_batch.updates);
613            cap_set.downgrade(remap_trace_batch.upper);
614        }
615    });
616
617    (remap_stream.as_collection(), button.press_on_drop())
618}
619
620/// Reclocks an `IntoTime` frontier stream into a `FromTime` frontier stream. This is used for the
621/// virtual (through persist) feedback edge so that we convert the `IntoTime` resumption frontier
622/// into the `FromTime` frontier that is used with the source's `OffsetCommiter`.
623fn reclock_committed_upper<G, FromTime>(
624    bindings: &Collection<G, FromTime, Diff>,
625    as_of: Antichain<G::Timestamp>,
626    committed_upper: &Stream<G, ()>,
627    id: GlobalId,
628    metrics: Arc<SourceMetrics>,
629) -> impl futures::stream::Stream<Item = Antichain<FromTime>> + 'static
630where
631    G: Scope,
632    G::Timestamp: Lattice + TotalOrder,
633    FromTime: SourceTimestamp,
634{
635    let (tx, rx) = watch::channel(Antichain::from_elem(FromTime::minimum()));
636    let scope = bindings.scope().clone();
637
638    let name = format!("ReclockCommitUpper({id})");
639    let mut builder = OperatorBuilderRc::new(name, scope);
640
641    let mut bindings = builder.new_input(&bindings.inner, Pipeline);
642    let _ = builder.new_input(committed_upper, Pipeline);
643
644    builder.build(move |_| {
645        // Remap bindings beyond the upper
646        use timely::progress::ChangeBatch;
647        let mut accepted_times: ChangeBatch<(G::Timestamp, FromTime)> = ChangeBatch::new();
648        // The upper frontier of the bindings
649        let mut upper = Antichain::from_elem(Timestamp::minimum());
650        // Remap bindings not beyond upper
651        let mut ready_times = VecDeque::new();
652        let mut source_upper = MutableAntichain::new();
653
654        move |frontiers| {
655            // Accept new bindings
656            while let Some((_, data)) = bindings.next() {
657                accepted_times.extend(data.drain(..).map(|(from, mut into, diff)| {
658                    into.advance_by(as_of.borrow());
659                    ((into, from), diff.into_inner())
660                }));
661            }
662            // Extract ready bindings
663            let new_upper = frontiers[0].frontier();
664            if PartialOrder::less_than(&upper.borrow(), &new_upper) {
665                upper = new_upper.to_owned();
666                // Drain consolidated accepted times not greater or equal to `upper` into `ready_times`.
667                // Retain accepted times greater or equal to `upper` in
668                let mut pending_times = std::mem::take(&mut accepted_times).into_inner();
669                // These should already be sorted, as part of `.into_inner()`, but sort defensively in case.
670                pending_times.sort_unstable_by(|a, b| a.0.cmp(&b.0));
671                for ((into, from), diff) in pending_times.drain(..) {
672                    if !upper.less_equal(&into) {
673                        ready_times.push_back((from, into, diff));
674                    } else {
675                        accepted_times.update((into, from), diff);
676                    }
677                }
678            }
679
680            // The received times only accumulate correctly for times beyond the as_of.
681            if as_of.iter().all(|t| !upper.less_equal(t)) {
682                let committed_upper = frontiers[1].frontier();
683                if as_of.iter().all(|t| !committed_upper.less_equal(t)) {
684                    // We have committed this source up until `committed_upper`. Because we have
685                    // required that IntoTime is a total order this will be either a singleton set
686                    // or the empty set.
687                    //
688                    // * Case 1: committed_upper is the empty set {}
689                    //
690                    // There won't be any future IntoTime timestamps that we will produce so we can
691                    // provide feedback to the source that it can forget about everything.
692                    //
693                    // * Case 2: committed_upper is a singleton set {t_next}
694                    //
695                    // We know that t_next cannot be the minimum timestamp because we have required
696                    // that all times of the as_of frontier are not beyond some time of
697                    // committed_upper. Therefore t_next has a predecessor timestamp t_prev.
698                    //
699                    // We don't know what remap[t_next] is yet, but we do know that we will have to
700                    // emit all source updates `u: remap[t_prev] <= time(u) <= remap[t_next]`.
701                    // Since `t_next` is the minimum undetermined timestamp and we know that t1 <=
702                    // t2 => remap[t1] <= remap[t2] we know that we will never need any source
703                    // updates `u: !(remap[t_prev] <= time(u))`.
704                    //
705                    // Therefore we can provide feedback to the source that it can forget about any
706                    // updates that are not beyond remap[t_prev].
707                    //
708                    // Important: We are *NOT* saying that the source can *compact* its data using
709                    // remap[t_prev] as the compaction frontier. If the source were to compact its
710                    // collection to remap[t_prev] we would lose the distinction between updates
711                    // that happened *at* t_prev versus updates that happened ealier and were
712                    // advanced to t_prev. If the source needs to communicate a compaction frontier
713                    // upstream then the specific source implementation needs to further adjust the
714                    // reclocked committed_upper and calculate a suitable compaction frontier in
715                    // the same way we adjust uppers of collections in the controller with the
716                    // LagWriteFrontier read policy.
717                    //
718                    // == What about IntoTime times that are general lattices?
719                    //
720                    // Reversing the upper for a general lattice is much more involved but it boils
721                    // down to computing the meet of all the times in `committed_upper` and then
722                    // treating that as `t_next` (I think). Until we need to deal with that though
723                    // we can just assume TotalOrder.
724                    let reclocked_upper = match committed_upper.as_option() {
725                        Some(t_next) => {
726                            let idx = ready_times.partition_point(|(_, t, _)| t < t_next);
727                            let updates = ready_times
728                                .drain(0..idx)
729                                .map(|(from_time, _, diff)| (from_time, diff));
730                            source_upper.update_iter(updates);
731                            // At this point source_upper contains all updates that are less than
732                            // t_next, which is equal to remap[t_prev]
733                            source_upper.frontier().to_owned()
734                        }
735                        None => Antichain::new(),
736                    };
737                    tx.send_replace(reclocked_upper);
738                }
739            }
740
741            metrics
742                .commit_upper_accepted_times
743                .set(u64::cast_from(accepted_times.len()));
744            metrics
745                .commit_upper_ready_times
746                .set(u64::cast_from(ready_times.len()));
747        }
748    });
749
750    WatchStream::from_changes(rx)
751}
752
753/// Synthesizes a probe stream that produces the frontier of the given progress stream at the given
754/// interval.
755///
756/// This is used as a fallback for sources that don't support probing the frontier of the upstream
757/// system.
758fn synthesize_probes<G>(
759    source_id: GlobalId,
760    progress: &Stream<G, Infallible>,
761    interval: Duration,
762    now_fn: NowFn,
763) -> Stream<G, Probe<G::Timestamp>>
764where
765    G: Scope,
766{
767    let scope = progress.scope();
768
769    let active_worker = usize::cast_from(source_id.hashed()) % scope.peers();
770    let is_active_worker = active_worker == scope.index();
771
772    let mut op = AsyncOperatorBuilder::new("synthesize_probes".into(), scope);
773    let (output, output_stream) = op.new_output();
774    let mut input = op.new_input_for(progress, Pipeline, &output);
775
776    op.build(|caps| async move {
777        if !is_active_worker {
778            return;
779        }
780
781        let [cap] = caps.try_into().expect("one capability per output");
782
783        let mut ticker = probe::Ticker::new(move || interval, now_fn.clone());
784
785        let minimum_frontier = Antichain::from_elem(Timestamp::minimum());
786        let mut frontier = minimum_frontier.clone();
787        loop {
788            tokio::select! {
789                event = input.next() => match event {
790                    Some(AsyncEvent::Progress(progress)) => frontier = progress,
791                    Some(AsyncEvent::Data(..)) => unreachable!(),
792                    None => break,
793                },
794                // We only report a probe if the source upper frontier is not the minimum frontier.
795                // This makes it so the first remap binding corresponds to the snapshot of the
796                // source, and because the first binding always maps to the minimum *target*
797                // frontier we guarantee that the source will never appear empty.
798                probe_ts = ticker.tick(), if frontier != minimum_frontier => {
799                    let probe = Probe {
800                        probe_ts,
801                        upstream_frontier: frontier.clone(),
802                    };
803                    output.give(&cap, probe);
804                }
805            }
806        }
807
808        let probe = Probe {
809            probe_ts: now_fn().into(),
810            upstream_frontier: Antichain::new(),
811        };
812        output.give(&cap, probe);
813    });
814
815    output_stream
816}