Committer pushes one update through store

crates/sui-indexer/README.md

@@ -32,7 +32,7 @@ See the [docs](https://docs.sui.io/guides/developer/getting-started/local-networ
 
 Start a local network using the `sui` binary:
 ```sh
-cargo run --bin sui -- start --with-faucet --force-regenesis 
+cargo run --bin sui -- start --with-faucet --force-regenesis
 ```
 
 If you want to run a local network with the indexer enabled (note that `libpq` is required), you can run the following command after following the steps in the next section to set up an indexer DB:
@@ -124,3 +124,9 @@ Note that you need an existing database for this to work. Using the DATABASE_URL
 # Change the RPC_CLIENT_URL to http://0.0.0.0:9000 to run indexer against local validator & fullnode
 cargo run --bin sui-indexer --features mysql-feature --no-default-features -- --db-url "<DATABASE_URL>" --rpc-client-url "https://fullnode.devnet.sui.io:443" --fullnode-sync-worker --reset-db
 ```
+
+### Extending the indexer
+
+To add a new table, run `diesel migration generate your_table_name`, and modify the newly created `up.sql` and `down.sql` files.
+
+You would apply the migration with `diesel migration run`, and run the script in `./scripts/generate_indexer_schema.sh` to update the `schema.rs` file.

crates/sui-indexer/migrations/pg/2024-09-12-213234_watermarks/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS watermarks;

crates/sui-indexer/migrations/pg/2024-09-12-213234_watermarks/up.sql

@@ -0,0 +1,32 @@
+CREATE TABLE watermarks
+(
+ -- The table governed by this watermark, i.e `epochs`, `checkpoints`, `transactions`.
+ entity TEXT NOT NULL,
+ -- Inclusive upper bound epoch this entity has data for. Committer updates this field. Pruner
+ -- uses this field for per-entity epoch-level retention, and is mostly useful for pruning
+ -- unpartitioned tables.
+ epoch_hi BIGINT NOT NULL,
+ -- Inclusive lower bound epoch this entity has data for. Pruner updates this field, and uses
+ -- this field in tandem with `epoch_hi` for per-entity epoch-level retention. This is mostly
+ -- useful for pruning unpartitioned tables.
+ epoch_lo BIGINT NOT NULL,
+ -- Inclusive upper bound checkpoint this entity has data for. Committer updates this field. All
+ -- data of this entity in the checkpoint must be persisted before advancing this watermark. The
+ -- committer or ingestion task refers to this on disaster recovery.
+ checkpoint_hi BIGINT NOT NULL,
+ -- Inclusive high watermark that the committer advances. For `checkpoints`, this represents the
+ -- checkpoint sequence number, for `transactions`, the transaction sequence number, etc.
+ reader_hi BIGINT NOT NULL,
+ -- Inclusive low watermark that the pruner advances. Data before this watermark is considered
+ -- pruned by a reader. The underlying data may still exist in the db instance.
+ reader_lo BIGINT NOT NULL,
+ -- Updated using the database's current timestamp when the pruner sees that some data needs to
+ -- be dropped. The pruner uses this column to determine whether to prune or wait long enough
+ -- that all in-flight reads complete or timeout before it acts on an updated watermark.
+ timestamp_ms BIGINT NOT NULL,
+ -- Column used by the pruner to track its true progress. Data at and below this watermark has
+ -- been truly pruned from the db, and should no longer exist. When recovering from a crash, the
+ -- pruner will consult this column to determine where to continue.
+ pruned_lo BIGINT,
+ PRIMARY KEY (entity)
+);

crates/sui-indexer/src/handlers/committer.rs

@@ -10,9 +10,11 @@ use tracing::{error, info};
 
 use sui_types::messages_checkpoint::CheckpointSequenceNumber;
 
+use crate::handlers::pruner::PrunableTable;
 use crate::metrics::IndexerMetrics;
 use crate::store::IndexerStore;
 use crate::types::IndexerResult;
+use strum::IntoEnumIterator;
 
 use super::{CheckpointDataToCommit, EpochToCommit};
 
@@ -57,6 +59,8 @@ where
  batch.push(checkpoint);
  next_checkpoint_sequence_number += 1;
  let epoch_number_option = epoch.as_ref().map(|epoch| epoch.new_epoch.epoch);
+ // The batch will consist of contiguous checkpoints and at most one epoch boundary at
+ // the end.
  if batch.len() == checkpoint_commit_batch_size || epoch.is_some() {
  commit_checkpoints(&state, batch, epoch, &metrics).await;
  batch = vec![];
@@ -79,6 +83,10 @@ where
  Ok(())
 }
 
+/// Writes indexed checkpoint data to the database, and then update watermark upper bounds and
+/// metrics. Expects `indexed_checkpoint_batch` to be non-empty, and contain contiguous checkpoints.
+/// There can be at most one epoch boundary at the end. If an epoch boundary is detected,
+/// epoch-partitioned tables must be advanced.
 // Unwrap: Caller needs to make sure indexed_checkpoint_batch is not empty
 #[instrument(skip_all, fields(
  first = indexed_checkpoint_batch.first().as_ref().unwrap().checkpoint.sequence_number,
@@ -130,7 +138,14 @@ async fn commit_checkpoints<S>(
  }
 
  let first_checkpoint_seq = checkpoint_batch.first().as_ref().unwrap().sequence_number;
- let last_checkpoint_seq = checkpoint_batch.last().as_ref().unwrap().sequence_number;
+ let (epoch_id, last_checkpoint_seq, last_tx_seq) = {
+ let checkpoint = checkpoint_batch.last().unwrap();
+ (
+ checkpoint.epoch,
+ checkpoint.sequence_number,
+ checkpoint.max_tx_sequence_number,
+ )
+ };
 
  let guard = metrics.checkpoint_db_commit_latency.start_timer();
  let tx_batch = tx_batch.into_iter().flatten().collect::<Vec<_>>();
@@ -189,7 +204,8 @@ async fn commit_checkpoints<S>(
 
  let is_epoch_end = epoch.is_some();
 
- // handle partitioning on epoch boundary
+ // On epoch boundary, we need to modify the existing partitions' upper bound, and introduce a
+ // new partition for incoming data for the upcoming epoch.
  if let Some(epoch_data) = epoch {
  state
  .advance_epoch(epoch_data)
@@ -212,6 +228,22 @@ async fn commit_checkpoints<S>(
  })
  .expect("Persisting data into DB should not fail.");
 
+ state
+ .update_watermarks_upper_bound(
+ PrunableTable::iter().collect(),
+ epoch_id,
+ last_checkpoint_seq,
+ last_tx_seq,
+ )
+ .await
+ .tap_err(|e| {
+ error!(
+ "Failed to update watermark upper bound with error: {}",
+ e.to_string()
+ );
+ })
+ .expect("Updating watermark upper bound in DB should not fail.");
+
  if is_epoch_end {
  // The epoch has advanced so we update the configs for the new protocol version, if it has changed.
  let chain_id = state

crates/sui-indexer/src/handlers/pruner.rs

@@ -72,6 +72,38 @@ pub enum PrunableTable {
  PrunerCpWatermark,
 }
 
+impl PrunableTable {
+ /// Given a committer's report of the latest written checkpoint and tx, return the value that
+ /// corresponds to the variant's unit to be used by readers.
+ pub fn map_to_reader_unit(&self, cp: u64, tx: u64) -> u64 {
+ match self {
+ PrunableTable::ObjectsHistory => cp,
+ PrunableTable::Checkpoints => cp,
+ PrunableTable::PrunerCpWatermark => cp,
+ PrunableTable::Events => tx,
+ PrunableTable::EventEmitPackage => tx,
+ PrunableTable::EventEmitModule => tx,
+ PrunableTable::EventSenders => tx,
+ PrunableTable::EventStructInstantiation => tx,
+ PrunableTable::EventStructModule => tx,
+ PrunableTable::EventStructName => tx,
+ PrunableTable::EventStructPackage => tx,
+ PrunableTable::Transactions => tx,
+ PrunableTable::TxAffectedAddresses => tx,
+ PrunableTable::TxAffectedObjects => tx,
+ PrunableTable::TxCallsPkg => tx,
+ PrunableTable::TxCallsMod => tx,
+ PrunableTable::TxCallsFun => tx,
+ PrunableTable::TxChangedObjects => tx,
+ PrunableTable::TxDigests => tx,
+ PrunableTable::TxInputObjects => tx,
+ PrunableTable::TxKinds => tx,
+ PrunableTable::TxRecipients => tx,
+ PrunableTable::TxSenders => tx,
+ }
+ }
+}
+
 impl Pruner {
  /// Instantiates a pruner with default retention and overrides. Pruner will merge these into a
  /// list of policies for each prunable table.

crates/sui-indexer/src/indexer.rs

@@ -114,6 +114,8 @@ impl Indexer {
 
  let mut exit_senders = vec![];
  let mut executors = vec![];
+ // Ingestion task watermarks are snapshotted once on indexer startup based on the
+ // corresponding watermark table before being handed off to the ingestion task.
  let progress_store = ShimIndexerProgressStore::new(vec![
  ("primary".to_string(), primary_watermark),
  ("object_snapshot".to_string(), object_snapshot_watermark),

crates/sui-indexer/src/metrics.rs

@@ -125,6 +125,7 @@ pub struct IndexerMetrics {
  pub checkpoint_db_commit_latency_tx_indices_chunks: Histogram,
  pub checkpoint_db_commit_latency_checkpoints: Histogram,
  pub checkpoint_db_commit_latency_epoch: Histogram,
+ pub checkpoint_db_commit_latency_watermarks: Histogram,
  pub thousand_transaction_avg_db_commit_latency: Histogram,
  pub object_db_commit_latency: Histogram,
  pub object_mutation_db_commit_latency: Histogram,
@@ -536,6 +537,13 @@ impl IndexerMetrics {
  registry,
  )
  .unwrap(),
+ checkpoint_db_commit_latency_watermarks: register_histogram_with_registry!(
+ "checkpoint_db_commit_latency_watermarks",
+ "Time spent committing watermarks",
+ DATA_INGESTION_LATENCY_SEC_BUCKETS.to_vec(),
+ registry,
+ )
+ .unwrap(),
  tokio_blocking_task_wait_latency: register_histogram_with_registry!(
  "tokio_blocking_task_wait_latency",
  "Time spent to wait for tokio blocking task pool",

crates/sui-indexer/src/models/mod.rs

@@ -12,3 +12,4 @@ pub mod packages;
 pub mod raw_checkpoints;
 pub mod transactions;
 pub mod tx_indices;
+pub mod watermarks;

crates/sui-indexer/src/models/watermarks.rs

@@ -0,0 +1,49 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::schema::watermarks::{self};
+use diesel::prelude::*;
+
+/// Represents a row in the `watermarks` table.
+#[derive(Queryable, Insertable, Default, QueryableByName)]
+#[diesel(table_name = watermarks, primary_key(entity))]
+pub struct StoredWatermark {
+ /// The table name of group of tables governed by this watermark, i.e `epochs`, `checkpoints`,
+ /// `transactions`.
+ pub entity: String,
+ /// Upper bound epoch range to enable per-entity epoch-level retention policy. Committer
+ /// advances this along with `high`.
+ pub epoch_hi: i64,
+ /// Lower bound epoch range to enable per-entity epoch-level retention policy. Pruner advances
+ /// this.
+ pub epoch_lo: i64,
+ pub checkpoint_hi: i64,
+ /// The inclusive high watermark that the committer advances.
+ pub reader_hi: i64,
+ /// The inclusive low watermark that the pruner advances. Data before this watermark is
+ /// considered pruned.
+ pub reader_lo: i64,
+ /// Pruner sets this, and uses this column to determine whether to prune or wait long enough
+ /// that all in-flight reads complete or timeout before it acts on an updated watermark.
+ pub timestamp_ms: i64,
+ /// Pruner updates this, and uses this when recovering from a crash to determine where to
+ /// continue pruning. Represents the latest watermark pruned, inclusive.
+ pub pruned_lo: Option<i64>,
+}
+
+impl StoredWatermark {
+ pub fn from_upper_bound_update(
+ entity: &str,
+ epoch_hi: u64,
+ checkpoint_hi: u64,
+ reader_hi: u64,
+ ) -> Self {
+ StoredWatermark {
+ entity: entity.to_string(),
+ epoch_hi: epoch_hi as i64,
+ checkpoint_hi: checkpoint_hi as i64,
+ reader_hi: reader_hi as i64,
+ ..StoredWatermark::default()
+ }
+ }
+}

crates/sui-indexer/src/schema.rs

@@ -368,6 +368,19 @@ diesel::table! {
  }
 }
 
+diesel::table! {
+ watermarks (entity) {
+ entity -> Text,
+ epoch_hi -> Int8,
+ epoch_lo -> Int8,
+ checkpoint_hi -> Int8,
+ reader_hi -> Int8,
+ reader_lo -> Int8,
+ timestamp_ms -> Int8,
+ pruned_lo -> Nullable<Int8>,
+ }
+}
+
 diesel::allow_tables_to_appear_in_same_query!(
  chain_identifier,
  checkpoints,
@@ -403,4 +416,5 @@ diesel::allow_tables_to_appear_in_same_query!(
  tx_kinds,
  tx_recipients,
  tx_senders,
+ watermarks,
 );

crates/sui-indexer/src/store/indexer_store.rs

@@ -6,6 +6,7 @@ use std::collections::BTreeMap;
 use async_trait::async_trait;
 
 use crate::errors::IndexerError;
+use crate::handlers::pruner::PrunableTable;
 use crate::handlers::{EpochToCommit, TransactionObjectChangesToCommit};
 use crate::models::display::StoredDisplay;
 use crate::models::obj_indices::StoredObjectVersion;
@@ -114,4 +115,12 @@ pub trait IndexerStore: Clone + Sync + Send + 'static {
  &self,
  checkpoints: Vec<StoredRawCheckpoint>,
  ) -> Result<(), IndexerError>;
+
+ async fn update_watermarks_upper_bound(
+ &self,
+ tables: Vec<PrunableTable>,
+ epoch: u64,
+ cp: u64,
+ tx: u64,
+ ) -> Result<(), IndexerError>;
 }