diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a3819c0c..59df52304 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,3 +35,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 instead of panicking the decoder. - SlateDB compaction skips a stored batch it cannot inflate, with a warning, instead of abandoning the whole maintenance pass. +- `DeleteRecords` is implemented on the libSQL, Turso and DynoStore storage + engines instead of panicking the connection task, and the PostgreSQL + implementation no longer fails at SQL-prepare time (it referenced columns + that don't exist in the current schema). All five storage engines + (PostgreSQL, libSQL, Turso, DynoStore, SlateDB) now validate the requested + offset identically: `-1` deletes up to the high watermark, an offset above + the high watermark or any other negative offset is rejected with + `OFFSET_OUT_OF_RANGE`, and an offset at or below the current log start is a + no-op. The record or batch holding the last committed offset is never + physically removed, even when deleting everything, so `ListOffsets(Latest)` + is unaffected by a `DeleteRecords` call. `ListOffsets(Earliest)` answers + the new log start on every storage engine. An unknown topic or an + out-of-range partition is reported per-partition and never fails a sibling + partition in the same request, and a partition that fails reports a low + watermark of `-1`, as Kafka does. On PostgreSQL each partition's log start + commits on its own, and removing the records below it follows as a separate + best-effort step: if that step fails or times out, the records stay below + the log start until the next storage maintenance run removes them. diff --git a/nisshi-broker/tests/it/delete_records.rs b/nisshi-broker/tests/it/delete_records.rs index d521fb3e1..6fbe0b3e3 100644 --- a/nisshi-broker/tests/it/delete_records.rs +++ b/nisshi-broker/tests/it/delete_records.rs @@ -12,56 +12,1062 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::init_tracing; -use nisshi_broker::Error; +use crate::common::{self, StorageType, alphanumeric_string, init_tracing, register_broker}; +use bytes::Bytes; +use nisshi_broker::Result; use nisshi_sans_io::{ - DeleteRecordsRequest, RequestInput, + DeleteRecordsRequest, ErrorCode, FetchRequest, IsolationLevel, ListOffset, NULL_TOPIC_ID, + RequestInput, + create_topics_request::CreatableTopic, delete_records_request::{DeleteRecordsPartition, DeleteRecordsTopic}, + delete_records_response::DeleteRecordsPartitionResult, + fetch_request::{FetchPartition, FetchTopic}, + record::{Record, inflated}, }; -use nisshi_storage::{DeleteRecordsService, StorageContainer}; +use nisshi_storage::{DeleteRecordsService, FetchService, ListOffsetResponse, Storage, Topition}; use rama::{Service, extensions::Extensions}; +use rand::{prelude::*, rng}; +use std::time::Duration; use tracing::debug; use url::Url; +use uuid::Uuid; -#[ignore] -#[tokio::test] -async fn delete_non_existent_records() -> Result<(), Error> { - let _guard = init_tracing()?; +async fn create_topic(sc: &G, num_partitions: i32) -> Result +where + G: Storage, +{ + let topic_name: String = alphanumeric_string(15); - let storage = StorageContainer::builder() - .cluster_id("nisshi") - .node_id(111) - .advertised_listener(Url::parse("tcp://localhost:9092")?) - .storage(Url::parse("memory://nisshi/")?) + let topic_id = sc + .create_topic( + CreatableTopic::default() + .name(topic_name.clone()) + .num_partitions(num_partitions) + .replication_factor(0) + .assignments(Some([].into())) + .configs(Some([].into())), + false, + ) + .await?; + debug!(?topic_id); + + Ok(topic_name) +} + +async fn produce_one(sc: &G, topition: &Topition) -> Result +where + G: Storage, +{ + let value = Bytes::copy_from_slice(alphanumeric_string(15).as_bytes()); + + let batch = inflated::Batch::builder() + .record(Record::builder().value(Some(value))) .build() + .and_then(TryInto::try_into)?; + + sc.produce(None, topition, batch).await.map_err(Into::into) +} + +async fn produce_n(sc: &G, topition: &Topition, n: i64) -> Result<()> +where + G: Storage, +{ + for _ in 0..n { + _ = produce_one(sc, topition).await?; + } + + Ok(()) +} + +/// Sends a single-topic, single-partition `DeleteRecords` request and +/// returns that one partition's result. +async fn delete_records_partition( + sc: &G, + topic: &str, + partition_index: i32, + offset: i64, +) -> Result +where + G: Storage + Clone, +{ + let response = DeleteRecordsService { + storage: sc.clone(), + } + .serve(RequestInput { + request: DeleteRecordsRequest::default().topics(Some( + [DeleteRecordsTopic::default() + .name(topic.into()) + .partitions(Some( + [DeleteRecordsPartition::default() + .partition_index(partition_index) + .offset(offset)] + .into(), + ))] + .into(), + )), + extensions: Extensions::default(), + }) + .await?; + + let topics = response.topics.unwrap_or_default(); + assert_eq!(1, topics.len()); + + let partitions = topics[0].partitions.clone().unwrap_or_default(); + assert_eq!(1, partitions.len()); + + Ok(partitions[0].clone()) +} + +/// Fetches from `fetch_offset` and returns the total number of records +/// returned. +async fn fetch_record_count(sc: &G, topition: &Topition, fetch_offset: i64) -> Result +where + G: Storage + Clone, +{ + let topics = [FetchTopic::default() + .topic(Some(topition.topic().to_string())) + .topic_id(Some(NULL_TOPIC_ID)) + .partitions(Some( + [FetchPartition::default() + .partition(topition.partition()) + .current_leader_epoch(Some(-1)) + .fetch_offset(fetch_offset) + .last_fetched_epoch(Some(-1)) + .log_start_offset(Some(-1)) + .partition_max_bytes(50 * 1024) + .replica_directory_id(None)] + .into(), + ))]; + + let fetch = FetchService { + storage: sc.clone(), + } + .serve(RequestInput { + request: FetchRequest::default() + .max_wait_ms(500) + .min_bytes(1) + .max_bytes(Some(50 * 1024)) + .isolation_level(Some((&IsolationLevel::ReadUncommitted).into())) + .topics(Some(topics.into())), + extensions: Extensions::default(), + }) + .await?; + + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(fetch.error_code.unwrap())? + ); + + Ok(fetch + .responses + .unwrap_or_default() + .iter() + .map(|response| { + response + .partitions + .as_deref() + .unwrap_or(&[]) + .iter() + .map(|partition| { + partition + .records + .as_ref() + .map(|records| { + records + .batches + .iter() + .map(|batch| batch.record_count as i64) + .sum::() + }) + .unwrap_or_default() + }) + .sum::() + }) + .sum::()) +} + +/// Produces `record_count` records, deletes everything up to a mid-stream +/// offset, and checks that the response, `offset_stage`, `ListOffsets` and +/// `Fetch` all agree on the new low watermark. +pub async fn delete_up_to_offset(cluster_id: Uuid, broker_id: i32, sc: G) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + let record_count = 6; + produce_n(&sc, &topition, record_count).await?; + + let cutoff = 3; + + let result = delete_records_partition(&sc, &topic_name, 0, cutoff).await?; + assert_eq!(ErrorCode::None, ErrorCode::try_from(result.error_code)?); + assert_eq!(cutoff, result.low_watermark); + + let stage = sc.offset_stage(&topition).await?; + assert_eq!(cutoff, stage.log_start()); + assert_eq!(record_count, stage.high_watermark()); + + let earliest = sc + .list_offsets( + IsolationLevel::ReadUncommitted, + &[(topition.clone(), ListOffset::Earliest)], + ) + .await?; + assert!(matches!( + earliest[..], + [(_, ListOffsetResponse { offset: Some(offset), .. })] if offset == cutoff + )); + + let latest = sc + .list_offsets( + IsolationLevel::ReadUncommitted, + &[(topition.clone(), ListOffset::Latest)], + ) .await?; + assert!(matches!( + latest[..], + [(_, ListOffsetResponse { offset: Some(offset), .. })] if offset == record_count + )); + + assert_eq!( + record_count - cutoff, + fetch_record_count(&sc, &topition, cutoff).await? + ); + + Ok(()) +} + +/// A cutoff that falls strictly inside a multi-record batch's range must +/// not take the whole batch with it. +/// +/// Every other scenario in this file produces single-record batches, so a +/// cutoff always lands exactly on a batch boundary. Here one batch holds +/// four records (offsets 0..=3) and the cutoff is 2: strictly above the +/// batch's base offset (0) and strictly below its end offset (3). The +/// correct rule deletes a batch only once its *end* offset is below the +/// cutoff; a backend that instead checks only a batch's *start* offset +/// would delete this whole batch -- including offsets 2 and 3, which must +/// still be visible -- because its base (0) is below the cutoff (2). +/// +/// A second, standalone record follows the batch so the straddled batch is +/// never the active segment: this test is purely about the end-vs-start +/// deletion rule, independent of the `high_watermark - 1` guard covered by +/// `delete_to_high_watermark_keeps_latest`. +pub async fn delete_offset_inside_a_batch_keeps_the_whole_batch( + cluster_id: Uuid, + broker_id: i32, + sc: G, +) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + let batch_record_count = 4i32; + + let mut builder = inflated::Batch::builder().last_offset_delta(batch_record_count - 1); + + for delta in 0..batch_record_count { + builder = builder.record( + Record::builder() + .offset_delta(delta) + .value(Some(Bytes::from(format!("record-{delta}")))), + ); + } + + let batch = builder.build().and_then(TryInto::try_into)?; + _ = sc.produce(None, &topition, batch).await?; + + // A standalone record at offset 4, after the batch. + _ = produce_one(&sc, &topition).await?; + + let record_count = i64::from(batch_record_count) + 1; + let cutoff = 2; + + let result = delete_records_partition(&sc, &topic_name, 0, cutoff).await?; + assert_eq!(ErrorCode::None, ErrorCode::try_from(result.error_code)?); + assert_eq!(cutoff, result.low_watermark); + + let stage = sc.offset_stage(&topition).await?; + assert_eq!(cutoff, stage.log_start()); + assert_eq!(record_count, stage.high_watermark()); + + let earliest = sc + .list_offsets( + IsolationLevel::ReadUncommitted, + &[(topition.clone(), ListOffset::Earliest)], + ) + .await?; + + assert!( + matches!( + earliest[..], + [(_, ListOffsetResponse { offset: Some(offset), .. })] if offset == cutoff + ), + "{earliest:?}" + ); + + // Every record at or after the cutoff must still be reachable, in + // order and without gaps. Fetching at an offset that falls inside a + // batch returns that batch whole (the client skips the records below + // the fetch offset) -- so on a batch-granular backend this legitimately + // hands back offsets 0 and 1 too, folded in with the batch that + // survived them. What it must never do is lose offsets 2 or 3: a + // start-based mutant deletes this whole batch merely because it + // *starts* (at 0) before the cutoff, which would leave only the + // standalone record at offset 4 -- a gap this loop catches. + let batches = sc + .fetch( + &topition, + cutoff, + 1, + 50 * 1024, + IsolationLevel::ReadUncommitted, + Duration::from_millis(500), + ) + .await? + .into_iter() + .try_fold(Vec::new(), |mut acc, batch| { + inflated::Batch::try_from(batch) + .map(|inflated| { + acc.push(inflated); + acc + }) + .map_err(nisshi_broker::Error::from) + })?; + + let mut expected = cutoff; + + for batch in &batches { + for record in &batch.records { + let offset = batch.base_offset + i64::from(record.offset_delta); + + // a whole batch can begin before the fetch offset on a + // batch-granular backend: the client skips the records below it + if offset < cutoff { + continue; + } + + assert_eq!(expected, offset, "fetch at cutoff {cutoff}"); + expected += 1; + } + } + + assert_eq!( + record_count, expected, + "fetch at cutoff {cutoff} is missing records" + ); + + Ok(()) +} + +/// `offset == -1` is the common case ("delete everything up to the high +/// watermark"). The record/batch holding `high_watermark - 1` is Kafka's +/// active segment and must survive regardless, so `Latest` -- and +/// `offset_stage`'s `high_watermark`, which `Latest` is ultimately built +/// from -- must be unaffected by even a "delete everything" request. +/// +/// `ListOffsets(Earliest)` answers the log start, the high watermark here, +/// and not the offset of the record that the delete keeps. +pub async fn delete_to_high_watermark_keeps_latest( + cluster_id: Uuid, + broker_id: i32, + sc: G, +) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + let record_count = 6; + produce_n(&sc, &topition, record_count).await?; + + let result = delete_records_partition(&sc, &topic_name, 0, -1).await?; + assert_eq!(ErrorCode::None, ErrorCode::try_from(result.error_code)?); + assert_eq!(record_count, result.low_watermark); + + let stage = sc.offset_stage(&topition).await?; + assert_eq!(record_count, stage.log_start()); + assert_eq!(record_count, stage.high_watermark()); + + let latest = sc + .list_offsets( + IsolationLevel::ReadUncommitted, + &[(topition.clone(), ListOffset::Latest)], + ) + .await?; + + assert!(matches!( + latest[..], + [(_, ListOffsetResponse { offset: Some(offset), .. })] if offset == record_count + )); + + let earliest = sc + .list_offsets( + IsolationLevel::ReadUncommitted, + &[(topition.clone(), ListOffset::Earliest)], + ) + .await?; + + assert!( + matches!( + earliest[..], + [(_, ListOffsetResponse { offset: Some(offset), .. })] if offset == record_count + ), + "{earliest:?}" + ); + + Ok(()) +} + +/// An offset above the high watermark is rejected and leaves the watermark +/// untouched. +pub async fn delete_above_high_watermark_is_rejected( + cluster_id: Uuid, + broker_id: i32, + sc: G, +) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + let record_count = 3; + produce_n(&sc, &topition, record_count).await?; + + let result = delete_records_partition(&sc, &topic_name, 0, record_count + 1).await?; + assert_eq!( + ErrorCode::OffsetOutOfRange, + ErrorCode::try_from(result.error_code)? + ); + assert_eq!(-1, result.low_watermark); + + let stage = sc.offset_stage(&topition).await?; + assert_eq!(0, stage.log_start()); + assert_eq!(record_count, stage.high_watermark()); + + Ok(()) +} + +/// Any negative offset other than `-1` is rejected. +pub async fn delete_negative_offset_other_than_minus_one_is_rejected( + cluster_id: Uuid, + broker_id: i32, + sc: G, +) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + produce_n(&sc, &topition, 3).await?; + + let result = delete_records_partition(&sc, &topic_name, 0, -2).await?; + assert_eq!( + ErrorCode::OffsetOutOfRange, + ErrorCode::try_from(result.error_code)? + ); + assert_eq!(-1, result.low_watermark); + + Ok(()) +} + +/// A request at or below the current log start is a no-op: it succeeds and +/// reports the *existing* low watermark, never moving it backward. +pub async fn delete_at_or_below_log_start_is_a_noop( + cluster_id: Uuid, + broker_id: i32, + sc: G, +) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + let record_count = 6; + produce_n(&sc, &topition, record_count).await?; - let service = DeleteRecordsService { - storage: storage.clone(), - }; + let first = delete_records_partition(&sc, &topic_name, 0, 3).await?; + assert_eq!(ErrorCode::None, ErrorCode::try_from(first.error_code)?); + assert_eq!(3, first.low_watermark); - let topic = "abcba"; + // Repeating the same cutoff, and asking for an even earlier one, are + // both no-ops that report the log start unchanged. + for offset in [3, 0] { + let repeat = delete_records_partition(&sc, &topic_name, 0, offset).await?; + assert_eq!(ErrorCode::None, ErrorCode::try_from(repeat.error_code)?); + assert_eq!(3, repeat.low_watermark); + } - let response = service - .serve(RequestInput { - request: DeleteRecordsRequest::default().topics(Some( - [DeleteRecordsTopic::default() - .name(topic.into()) + let stage = sc.offset_stage(&topition).await?; + assert_eq!(3, stage.log_start()); + + Ok(()) +} + +/// An unknown topic and an out-of-range partition each get their own +/// per-partition `UnknownTopicOrPartition`, while a sibling valid partition +/// in the *same* request still succeeds -- proving a per-partition problem +/// never fails the whole request. +pub async fn delete_unknown_topic_or_partition_does_not_fail_siblings( + cluster_id: Uuid, + broker_id: i32, + sc: G, +) -> Result<()> +where + G: Storage + Clone, +{ + register_broker(cluster_id, broker_id, &sc).await?; + + let topic_name = create_topic(&sc, 1).await?; + let topition = Topition::new(topic_name.clone(), 0); + + let record_count = 4; + produce_n(&sc, &topition, record_count).await?; + + let unknown_topic: String = alphanumeric_string(15); + + let response = DeleteRecordsService { + storage: sc.clone(), + } + .serve(RequestInput { + request: DeleteRecordsRequest::default().topics(Some( + [ + DeleteRecordsTopic::default() + .name(unknown_topic.clone()) .partitions(Some( [DeleteRecordsPartition::default() - .offset(32123) - .partition_index(6)] + .partition_index(0) + .offset(0)] .into(), - ))] - .into(), - )), - extensions: Extensions::default(), - }) - .await - .inspect(|response| debug!(?response))?; + )), + DeleteRecordsTopic::default() + .name(topic_name.clone()) + .partitions(Some( + [ + DeleteRecordsPartition::default() + .partition_index(1) + .offset(0), + DeleteRecordsPartition::default() + .partition_index(0) + .offset(2), + ] + .into(), + )), + ] + .into(), + )), + extensions: Extensions::default(), + }) + .await?; let topics = response.topics.unwrap_or_default(); - assert_eq!(1, topics.len()); + assert_eq!(2, topics.len()); + + let unknown_topic_result = topics + .iter() + .find(|topic| topic.name == unknown_topic) + .expect("unknown topic result"); + let unknown_topic_partitions = unknown_topic_result.partitions.clone().unwrap_or_default(); + assert_eq!(1, unknown_topic_partitions.len()); + assert_eq!( + ErrorCode::UnknownTopicOrPartition, + ErrorCode::try_from(unknown_topic_partitions[0].error_code)? + ); + assert_eq!(-1, unknown_topic_partitions[0].low_watermark); + + let known_topic_result = topics + .iter() + .find(|topic| topic.name == topic_name) + .expect("known topic result"); + let known_topic_partitions = known_topic_result.partitions.clone().unwrap_or_default(); + assert_eq!(2, known_topic_partitions.len()); + + let out_of_range_partition = known_topic_partitions + .iter() + .find(|partition| partition.partition_index == 1) + .expect("out of range partition result"); + assert_eq!( + ErrorCode::UnknownTopicOrPartition, + ErrorCode::try_from(out_of_range_partition.error_code)? + ); + assert_eq!(-1, out_of_range_partition.low_watermark); + + let valid_partition = known_topic_partitions + .iter() + .find(|partition| partition.partition_index == 0) + .expect("valid sibling partition result"); + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(valid_partition.error_code)? + ); + assert_eq!(2, valid_partition.low_watermark); + + let stage = sc.offset_stage(&topition).await?; + assert_eq!(2, stage.log_start()); Ok(()) } + +#[cfg(feature = "postgres")] +mod pg { + use super::*; + use nisshi_storage::ArcDynStorage; + + async fn storage_container( + cluster: impl Into + Clone, + node: i32, + ) -> Result { + common::storage_container( + StorageType::Postgres, + cluster, + node, + Url::parse("tcp://127.0.0.1/")?, + None, + ) + .await + } + + #[tokio::test] + async fn delete_up_to_offset() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_up_to_offset( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_offset_inside_a_batch_keeps_the_whole_batch() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_offset_inside_a_batch_keeps_the_whole_batch( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_to_high_watermark_keeps_latest() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_to_high_watermark_keeps_latest( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_above_high_watermark_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_above_high_watermark_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_negative_offset_other_than_minus_one_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_negative_offset_other_than_minus_one_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_at_or_below_log_start_is_a_noop() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_at_or_below_log_start_is_a_noop( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_unknown_topic_or_partition_does_not_fail_siblings() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_unknown_topic_or_partition_does_not_fail_siblings( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } +} + +#[cfg(feature = "dynostore")] +mod in_memory { + use super::*; + use nisshi_storage::ArcDynStorage; + + async fn storage_container( + cluster: impl Into + Clone, + node: i32, + ) -> Result { + common::storage_container( + StorageType::InMemory, + cluster, + node, + Url::parse("tcp://127.0.0.1/")?, + None, + ) + .await + } + + #[tokio::test] + async fn delete_up_to_offset() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_up_to_offset( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_offset_inside_a_batch_keeps_the_whole_batch() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_offset_inside_a_batch_keeps_the_whole_batch( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_to_high_watermark_keeps_latest() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_to_high_watermark_keeps_latest( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_above_high_watermark_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_above_high_watermark_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_negative_offset_other_than_minus_one_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_negative_offset_other_than_minus_one_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_at_or_below_log_start_is_a_noop() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_at_or_below_log_start_is_a_noop( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_unknown_topic_or_partition_does_not_fail_siblings() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_unknown_topic_or_partition_does_not_fail_siblings( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } +} + +#[cfg(feature = "libsql")] +mod lite { + use super::*; + use nisshi_storage::ArcDynStorage; + + async fn storage_container( + cluster: impl Into + Clone, + node: i32, + ) -> Result { + common::storage_container( + StorageType::Lite, + cluster, + node, + Url::parse("tcp://127.0.0.1/")?, + None, + ) + .await + } + + #[tokio::test] + async fn delete_up_to_offset() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_up_to_offset( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_offset_inside_a_batch_keeps_the_whole_batch() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_offset_inside_a_batch_keeps_the_whole_batch( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_to_high_watermark_keeps_latest() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_to_high_watermark_keeps_latest( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_above_high_watermark_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_above_high_watermark_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_negative_offset_other_than_minus_one_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_negative_offset_other_than_minus_one_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_at_or_below_log_start_is_a_noop() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_at_or_below_log_start_is_a_noop( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_unknown_topic_or_partition_does_not_fail_siblings() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_unknown_topic_or_partition_does_not_fail_siblings( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } +} + +#[cfg(feature = "slatedb")] +mod slatedb { + use super::*; + use nisshi_storage::ArcDynStorage; + + async fn storage_container( + cluster: impl Into + Clone, + node: i32, + ) -> Result { + common::storage_container( + StorageType::SlateDb, + cluster, + node, + Url::parse("tcp://127.0.0.1/")?, + None, + ) + .await + } + + #[tokio::test] + async fn delete_up_to_offset() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_up_to_offset( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_offset_inside_a_batch_keeps_the_whole_batch() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_offset_inside_a_batch_keeps_the_whole_batch( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_to_high_watermark_keeps_latest() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_to_high_watermark_keeps_latest( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_above_high_watermark_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_above_high_watermark_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_negative_offset_other_than_minus_one_is_rejected() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_negative_offset_other_than_minus_one_is_rejected( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_at_or_below_log_start_is_a_noop() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_at_or_below_log_start_is_a_noop( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[tokio::test] + async fn delete_unknown_topic_or_partition_does_not_fail_siblings() -> Result<()> { + let _guard = init_tracing()?; + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + super::delete_unknown_topic_or_partition_does_not_fail_siblings( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } +} diff --git a/nisshi-storage-dynostore/src/dynostore.rs b/nisshi-storage-dynostore/src/dynostore.rs index 1dfd45e01..c2eebc518 100644 --- a/nisshi-storage-dynostore/src/dynostore.rs +++ b/nisshi-storage-dynostore/src/dynostore.rs @@ -38,7 +38,7 @@ use nisshi_sans_io::{ create_topics_request::{CreatableTopic, CreatableTopicConfig}, delete_groups_response::DeletableGroupResult, delete_records_request::DeleteRecordsTopic, - delete_records_response::DeleteRecordsTopicResult, + delete_records_response::{DeleteRecordsPartitionResult, DeleteRecordsTopicResult}, describe_cluster_response::DescribeClusterBroker, describe_configs_response::{DescribeConfigsResourceResult, DescribeConfigsResult}, describe_topic_partitions_response::{ @@ -57,10 +57,11 @@ use nisshi_schema::{ lake::{House, LakeHouse as _}, }; use nisshi_storage::{ - BrokerRegistrationRequest, Error, GroupDetail, ListOffsetResponse, METER, MetadataResponse, - NamedGroupDetail, OffsetCommitRequest, OffsetStage, ProducerIdResponse, Result, - ScramCredential, Storage, TopicId, Topition, TxnAddPartitionsRequest, TxnAddPartitionsResponse, - TxnOffsetCommitRequest, TxnState, UpdateError, Version, + BrokerRegistrationRequest, DELETE_RECORDS_INVALID_LOW_WATERMARK, Error, GroupDetail, + ListOffsetResponse, METER, MetadataResponse, NamedGroupDetail, OffsetCommitRequest, + OffsetStage, ProducerIdResponse, Result, ScramCredential, Storage, TopicId, Topition, + TxnAddPartitionsRequest, TxnAddPartitionsResponse, TxnOffsetCommitRequest, TxnState, + UpdateError, Version, delete_records_cutoff, }; use object_store::{ Attribute, AttributeValue, Attributes, CopyOptions, DynObjectStore, GetOptions, GetResult, @@ -430,6 +431,93 @@ impl DynoStore { .map_err(Into::into) } + /// Removes the batch objects of one partition that lie entirely below + /// `cutoff`, after `DeleteRecords` has committed `cutoff` as the low + /// watermark. + /// + /// Each failure is logged and skipped, instead of returned, because the + /// watermark advance has already taken effect: an error here would report + /// a failed request for a delete that succeeded. + /// + /// An object is named by the base offset of its batch, and the name does + /// not hold the end offset. A batch ends before the base offset of the + /// next batch, so this deletes an object when the next object's base + /// offset is at or below `cutoff`, without reading either object. The + /// last object has no successor, so this never deletes it. That object + /// holds `high_watermark - 1`, which `ListOffsets(Latest)` reads. + /// + /// The listing covers this partition's `records/` prefix only, so that a + /// mistake here cannot remove the objects of another partition. + async fn delete_records_batches(&self, topition: &Topition, cutoff: i64) { + if cutoff <= 0 { + return; + } + + let prefix = Path::from(format!( + "clusters/{}/topics/{}/partitions/{:0>10}/records/", + self.cluster, topition.topic, topition.partition + )); + + let mut objects = match self + .object_store + .list(Some(&prefix)) + .map_ok(|meta| meta.location) + .try_collect::>() + .await + { + Ok(locations) => locations + .into_iter() + .filter_map(|location| { + let base_offset = location + .parts() + .next_back() + .and_then(|name| i64::from_str(name.as_ref().get(0..20)?).ok())?; + + Some((location, base_offset)) + }) + .collect::>(), + + Err(error) => { + error!(?error, ?topition, cutoff); + return; + } + }; + + objects.sort_by_key(|(_, base_offset)| *base_offset); + + let expired = objects + .windows(2) + .filter(|window| window[1].1 <= cutoff) + .map(|window| Ok::<_, object_store::Error>(window[0].0.clone())) + .collect::>(); + + let mut deleted = self + .object_store + .delete_stream(futures::stream::iter(expired).boxed()); + + while let Some(outcome) = deleted.next().await { + if let Err(error) = outcome { + error!(?error, ?topition, cutoff); + } + } + } + + /// Returns the low watermark of one partition: its log start offset. + async fn log_start(&self, topition: &Topition) -> Result { + let watermark = self.watermarks.lock().map(|mut locked| { + locked + .entry(topition.to_owned()) + .or_insert_with(|| OptiCon::::new(self.cluster.as_str(), topition)) + .to_owned() + })?; + + watermark + .with(&self.object_store, |watermark| { + Ok(watermark.low.unwrap_or(0)) + }) + .await + } + async fn get(&self, location: &Path) -> Result<(V, Version)> where V: DeserializeOwned, @@ -634,9 +722,108 @@ impl Storage for DynoStore { async fn delete_records( &self, - _topics: &[DeleteRecordsTopic], + topics: &[DeleteRecordsTopic], ) -> Result> { - todo!() + debug!(?topics); + + let mut responses = vec![]; + + for topic in topics { + let mut partition_responses = vec![]; + + if let Some(ref partitions) = topic.partitions { + let metadata = self + .topic_metadata(&TopicId::Name(topic.name.clone())) + .await?; + + for partition in partitions { + let (error_code, low_watermark) = match &metadata { + None => ( + ErrorCode::UnknownTopicOrPartition, + DELETE_RECORDS_INVALID_LOW_WATERMARK, + ), + + Some(metadata) + if partition.partition_index < 0 + || partition.partition_index >= metadata.topic.num_partitions => + { + ( + ErrorCode::UnknownTopicOrPartition, + DELETE_RECORDS_INVALID_LOW_WATERMARK, + ) + } + + Some(_) => { + let topition = + Topition::new(topic.name.as_str(), partition.partition_index); + + let watermark = self.watermarks.lock().map(|mut locked| { + locked + .entry(topition.to_owned()) + .or_insert_with(|| { + OptiCon::::new(self.cluster.as_str(), &topition) + }) + .to_owned() + })?; + + // Resolve and validate the requested offset against the + // *current* watermark, and commit it atomically: `with_mut` + // re-runs this closure against the latest value on a + // concurrent-write conflict (e.g. a produce racing this + // delete), so the cutoff is always computed from the value + // actually being persisted. + let (error_code, cutoff) = watermark + .with_mut(&self.object_store, |w| { + let stage = OffsetStage { + last_stable: w.high.unwrap_or(0), + high_watermark: w.high.unwrap_or(0), + log_start: w.low.unwrap_or(0), + }; + + match delete_records_cutoff(partition.offset, &stage) { + Err(error_code) => { + Ok((error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK)) + } + + Ok(cutoff) => { + w.low = Some(cutoff); + Ok((ErrorCode::None, cutoff)) + } + } + }) + .await?; + + if error_code == ErrorCode::None { + // Best-effort physical cleanup of batches now fully + // below the new low watermark. Failing to reclaim + // storage here does not undo the watermark advance + // above: the logical delete has already taken effect, + // same as Kafka's own segment reclaim lagging behind + // a logStartOffset bump. + self.delete_records_batches(&topition, cutoff).await; + } + + (error_code, cutoff) + } + }; + + partition_responses.push( + DeleteRecordsPartitionResult::default() + .partition_index(partition.partition_index) + .low_watermark(low_watermark) + .error_code(error_code.into()), + ); + } + } + + responses.push( + DeleteRecordsTopicResult::default() + .name(topic.name.clone()) + .partitions(Some(partition_responses)), + ); + } + + Ok(responses) } async fn delete_topic(&self, topic: &TopicId) -> Result { @@ -1319,6 +1506,32 @@ impl Storage for DynoStore { debug!(?candidate); + if offset_request == &ListOffset::Earliest { + let log_start = self.log_start(topition).await?; + + // Kafka answers `Earliest` with the log start offset. An object + // that begins below the log start can remain after + // `DeleteRecords`, so the base offset of the oldest object can + // be below the log start. + let base_offset = candidate + .as_ref() + .and_then(|found| found.location.parts().next_back()) + .and_then(|name| i64::from_str(name.as_ref().get(0..20)?).ok()); + + if base_offset.is_none_or(|base_offset| base_offset < log_start) { + responses.push(( + topition.to_owned(), + ListOffsetResponse { + error_code: ErrorCode::None, + offset: Some(log_start), + ..Default::default() + }, + )); + + continue; + } + } + if let Some(ref found) = candidate { let Some(offset) = found.location.parts().next_back() else { continue; diff --git a/nisshi-storage-slatedb/src/storage.rs b/nisshi-storage-slatedb/src/storage.rs index e06967b27..efcb4912b 100644 --- a/nisshi-storage-slatedb/src/storage.rs +++ b/nisshi-storage-slatedb/src/storage.rs @@ -50,10 +50,11 @@ use nisshi_sans_io::{ }; use nisshi_schema::lake::LakeHouse as _; use nisshi_storage::{ - BrokerRegistrationRequest, Error, GroupDetail, ListOffsetResponse, MetadataResponse, - NamedGroupDetail, OffsetCommitRequest, OffsetStage, ProducerIdResponse, Result, - ScramCredential, Storage, TopicId, Topition, TxnAddPartitionsRequest, TxnAddPartitionsResponse, - TxnOffsetCommitRequest, TxnState, UpdateError, Version, + BrokerRegistrationRequest, DELETE_RECORDS_INVALID_LOW_WATERMARK, Error, GroupDetail, + ListOffsetResponse, MetadataResponse, NamedGroupDetail, OffsetCommitRequest, OffsetStage, + ProducerIdResponse, Result, ScramCredential, Storage, TopicId, Topition, + TxnAddPartitionsRequest, TxnAddPartitionsResponse, TxnOffsetCommitRequest, TxnState, + UpdateError, Version, delete_records_cutoff, }; use serde::Serialize; use tracing::{debug, warn}; @@ -515,42 +516,11 @@ impl Storage for Engine { if partition.partition_index < 0 || partition.partition_index >= metadata.topic.num_partitions { - (ErrorCode::UnknownTopicOrPartition, 0) + ( + ErrorCode::UnknownTopicOrPartition, + DELETE_RECORDS_INVALID_LOW_WATERMARK, + ) } else { - // Delete batches below the specified offset - let batch_prefix = postcard::to_stdvec(&BatchKeyPrefix::new( - metadata.id, - partition.partition_index, - ))?; - let scan_start = postcard::to_stdvec(&BatchKey::scan_from( - metadata.id, - partition.partition_index, - 0, - ))?; - - let mut scan = self.db.scan(scan_start..).await?; - while let Some(kv) = scan.next().await? { - if !kv.key.starts_with(&batch_prefix) { - break; - } - - let batch_key: BatchKey = match postcard::from_bytes(&kv.key) { - Ok(key) => key, - Err(_) => continue, - }; - - // Delete batches with offset < specified offset - if batch_key.offset >= partition.offset { - break; - } - - tx.delete(&kv.key)?; - } - - // The new low watermark is the requested offset - let new_low_watermark = partition.offset; - - // Update the watermark let watermark_key = postcard::to_stdvec(&WatermarkKey::new( metadata.id, partition.partition_index, @@ -565,20 +535,89 @@ impl Storage for Engine { }, )?; - watermark.low = Some(new_low_watermark); + let high_watermark = watermark.high.unwrap_or(0); + let log_start = watermark.low.unwrap_or(0); - // Remove timestamps before the new low watermark - if let Some(ref mut timestamps) = watermark.timestamps { - timestamps.retain(|_, offset| *offset >= new_low_watermark); - } + let stage = OffsetStage { + last_stable: high_watermark, + high_watermark, + log_start, + }; - let watermark_value = postcard::to_stdvec(&watermark)?; - tx.put(&watermark_key, watermark_value)?; + match delete_records_cutoff(partition.offset, &stage) { + Err(error_code) => { + (error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK) + } - (ErrorCode::None, new_low_watermark) + Ok(cutoff) => { + if cutoff > log_start { + // A batch key holds the base offset of the batch, + // and not its end offset. A batch ends before the + // base offset of the next batch, so this deletes a + // batch when the next batch's base offset is at or + // below the cutoff, without decoding either batch. + // A batch that straddles the cutoff stays whole. + // The last batch has no successor, so this never + // deletes it. That batch holds `high_watermark - 1`. + let batch_prefix = + postcard::to_stdvec(&BatchKeyPrefix::new( + metadata.id, + partition.partition_index, + ))?; + let scan_start = + postcard::to_stdvec(&BatchKey::scan_from( + metadata.id, + partition.partition_index, + 0, + ))?; + + let mut previous: Option = None; + + let mut scan = self.db.scan(scan_start..).await?; + while let Some(kv) = scan.next().await? { + if !kv.key.starts_with(&batch_prefix) { + break; + } + + let Ok(batch_key) = + postcard::from_bytes::(&kv.key) + else { + continue; + }; + + if batch_key.offset > cutoff { + break; + } + + if let Some(key) = previous.replace(kv.key) { + tx.delete(&key)?; + } + + if batch_key.offset == cutoff { + break; + } + } + } + + watermark.low = Some(cutoff); + + // Remove timestamps before the new low watermark + if let Some(ref mut timestamps) = watermark.timestamps { + timestamps.retain(|_, offset| *offset >= cutoff); + } + + let watermark_value = postcard::to_stdvec(&watermark)?; + tx.put(&watermark_key, watermark_value)?; + + (ErrorCode::None, cutoff) + } + } } } else { - (ErrorCode::UnknownTopicOrPartition, 0) + ( + ErrorCode::UnknownTopicOrPartition, + DELETE_RECORDS_INVALID_LOW_WATERMARK, + ) }; partition_results.push( diff --git a/nisshi-storage-slatedb/src/tests.rs b/nisshi-storage-slatedb/src/tests.rs index b26e0d37e..79a781a95 100644 --- a/nisshi-storage-slatedb/src/tests.rs +++ b/nisshi-storage-slatedb/src/tests.rs @@ -50,6 +50,18 @@ async fn create_test_engine() -> Engine { ) } +/// A single-record batch, built through the encode path, so that it decodes +/// back to the record it holds. +fn simple_batch() -> Batch { + use nisshi_sans_io::record::{Record, inflated}; + + inflated::Batch::builder() + .record(Record::builder().value(Some(Bytes::from_static(b"value")))) + .build() + .and_then(Batch::try_from) + .unwrap() +} + // ========== Unique Error Case Tests ========== #[tokio::test] @@ -341,22 +353,7 @@ async fn test_delete_records() { let topition = Topition::new("delete-records-topic", 0); // Produce some data - let batch = Batch { - base_offset: 0, - batch_length: 0, - partition_leader_epoch: 0, - magic: 2, - crc: 0, - attributes: 0, - last_offset_delta: 0, - base_timestamp: 1000, - max_timestamp: 1000, - producer_id: -1, - producer_epoch: -1, - base_sequence: -1, - record_count: 1, - record_data: Bytes::new(), - }; + let batch = simple_batch(); for _ in 0..5 { let _ = engine @@ -397,6 +394,101 @@ async fn test_delete_records() { assert_eq!(3, stage.log_start); } +/// `offset == -1` ("delete everything up to the high watermark") must never +/// physically remove the batch holding `high_watermark - 1`: it's Kafka's +/// active segment. The public API can't observe this directly -- `Latest` +/// is read straight from the stored watermark, not by scanning batches, so +/// every public-API test still passes even with that protection deleted +/// entirely. This test builds the `Engine` from a `Db` handle it keeps for +/// itself, so it can scan the batch-key prefix directly afterward and +/// confirm the one physical object that must survive actually does. +#[tokio::test] +async fn test_delete_records_to_high_watermark_keeps_one_physical_batch() { + use super::types::{BatchKey, BatchKeyPrefix}; + use nisshi_sans_io::delete_records_request::{DeleteRecordsPartition, DeleteRecordsTopic}; + + let object_store = Arc::new(InMemory::new()); + let db = Arc::new( + Db::open("test.slatedb", object_store) + .await + .expect("Failed to open SlateDB"), + ); + + let engine = Engine::new( + "test-cluster", + 1, + Url::parse("tcp://localhost:9092").unwrap(), + db.clone(), + ); + + let topic = CreatableTopic::default() + .name("protect-active-batch-topic".into()) + .num_partitions(1) + .replication_factor(1); + let topic_id = engine.create_topic(topic, false).await.unwrap(); + + let topition = Topition::new("protect-active-batch-topic", 0); + + // Five single-record batches: offsets 0..=4, each its own physical + // batch object (unlike `test_delete_records` above, which only needs + // log_start/low_watermark to move, not an exact physical object count). + for _ in 0..5 { + let batch = simple_batch(); + let _ = engine.produce(None, &topition, batch).await.unwrap(); + } + + async fn batch_offsets(db: &Db, topic_id: uuid::Uuid, partition: i32) -> Vec { + let prefix = postcard::to_stdvec(&BatchKeyPrefix::new(topic_id, partition)).unwrap(); + let scan_start = postcard::to_stdvec(&BatchKey::scan_from(topic_id, partition, 0)).unwrap(); + + let mut scan = db.scan(scan_start..).await.unwrap(); + let mut offsets = vec![]; + + while let Some(kv) = scan.next().await.unwrap() { + if !kv.key.starts_with(&prefix) { + break; + } + + offsets.push(postcard::from_bytes::(&kv.key).unwrap().offset); + } + + offsets + } + + assert_eq!( + vec![0, 1, 2, 3, 4], + batch_offsets(&db, topic_id, 0).await, + "setup: expected one physical batch object per produced record" + ); + + let delete_request = vec![ + DeleteRecordsTopic::default() + .name("protect-active-batch-topic".into()) + .partitions(Some(vec![ + DeleteRecordsPartition::default() + .partition_index(0) + .offset(-1), + ])), + ]; + + let results = engine.delete_records(&delete_request).await.unwrap(); + let partitions = results[0].partitions.as_ref().unwrap(); + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(partitions[0].error_code).unwrap() + ); + assert_eq!(5, partitions[0].low_watermark); + + // Only the batch at offset 4, which holds high_watermark - 1, remains. + // Every batch ends below a cutoff equal to the high watermark, so this + // batch survives only because the delete keeps the last batch. + assert_eq!( + vec![4], + batch_offsets(&db, topic_id, 0).await, + "the active batch holding high_watermark - 1 must survive a -1 delete" + ); +} + #[tokio::test] async fn test_fetch_with_min_bytes() { let engine = create_test_engine().await; @@ -938,22 +1030,7 @@ async fn test_list_offsets_earliest_after_delete_records() { let topition = Topition::new("earliest-topic", 0); - let batch = Batch { - base_offset: 0, - batch_length: 0, - partition_leader_epoch: 0, - magic: 2, - crc: 0, - attributes: 0, - last_offset_delta: 0, - base_timestamp: 1000, - max_timestamp: 1000, - producer_id: -1, - producer_epoch: -1, - base_sequence: -1, - record_count: 1, - record_data: Bytes::new(), - }; + let batch = simple_batch(); for _ in 0..5 { let _ = engine diff --git a/nisshi-storage-sql/src/limbo.rs b/nisshi-storage-sql/src/limbo.rs index ed123db50..c9f8137e9 100644 --- a/nisshi-storage-sql/src/limbo.rs +++ b/nisshi-storage-sql/src/limbo.rs @@ -37,7 +37,7 @@ use nisshi_sans_io::{ create_topics_request::CreatableTopic, delete_groups_response::DeletableGroupResult, delete_records_request::DeleteRecordsTopic, - delete_records_response::DeleteRecordsTopicResult, + delete_records_response::{DeleteRecordsPartitionResult, DeleteRecordsTopicResult}, describe_cluster_response::DescribeClusterBroker, describe_configs_response::{DescribeConfigsResourceResult, DescribeConfigsResult}, describe_topic_partitions_response::{ @@ -58,10 +58,11 @@ use nisshi_schema::{ redact_url, }; use nisshi_storage::{ - BrokerRegistrationRequest, Error, GroupDetail, ListOffsetRequest, ListOffsetResponse, METER, - MetadataResponse, NamedGroupDetail, OffsetCommitRequest, OffsetStage, ProducerIdResponse, - Result, ScramCredential, Storage, TopicId, Topition, TxnAddPartitionsRequest, - TxnAddPartitionsResponse, TxnOffsetCommitRequest, TxnState, UpdateError, Version, + BrokerRegistrationRequest, DELETE_RECORDS_INVALID_LOW_WATERMARK, Error, GroupDetail, + ListOffsetRequest, ListOffsetResponse, METER, MetadataResponse, NamedGroupDetail, + OffsetCommitRequest, OffsetStage, ProducerIdResponse, Result, ScramCredential, Storage, + TopicId, Topition, TxnAddPartitionsRequest, TxnAddPartitionsResponse, TxnOffsetCommitRequest, + TxnState, UpdateError, Version, delete_records_cutoff, }; use opentelemetry::{ KeyValue, @@ -1232,7 +1233,105 @@ impl Storage for Engine { topics: &[DeleteRecordsTopic], ) -> Result> { debug!(?topics); - todo!() + + let mut connection = self.connection().await?; + let tx = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .await?; + + let mut responses = vec![]; + + for topic in topics { + let mut partition_responses = vec![]; + + if let Some(ref partitions) = topic.partitions { + for partition in partitions { + let topition = Topition::new(topic.name.as_str(), partition.partition_index); + + let (error_code, low_watermark) = + match self.watermark_select_for_update(&topition, &tx).await { + Err(Error::Api(error_code)) => { + (error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK) + } + Err(err) => return Err(err), + + Ok((low, high)) => { + let stage = OffsetStage { + last_stable: high.unwrap_or_default(), + high_watermark: high.unwrap_or_default(), + log_start: low.unwrap_or_default(), + }; + + match delete_records_cutoff(partition.offset, &stage) { + Err(error_code) => { + (error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK) + } + + Ok(cutoff) => { + // This keeps the record at `high_watermark - 1`, because + // `list_latest_offset_committed.sql` and + // `list_latest_offset_uncommitted.sql` answer `Latest` from + // that record. The watermark still advances to `cutoff`. + let physical_cutoff = cutoff.min(stage.high_watermark - 1); + + if physical_cutoff > stage.log_start { + _ = self + .prepare_execute( + &tx, + &sql_lookup("record_delete_by_offset.sql")?, + ( + self.cluster.as_str(), + topic.name.as_str(), + partition.partition_index, + physical_cutoff, + ), + ) + .await + .inspect_err(|err| { + error!(?err, ?topition, physical_cutoff) + })?; + } + + _ = self + .prepare_execute( + &tx, + &sql_lookup("watermark_update.sql")?, + ( + self.cluster.as_str(), + topic.name.as_str(), + partition.partition_index, + cutoff, + high.unwrap_or_default(), + ), + ) + .await + .inspect_err(|err| error!(?err, ?topition))?; + + (ErrorCode::None, cutoff) + } + } + } + }; + + partition_responses.push( + DeleteRecordsPartitionResult::default() + .partition_index(partition.partition_index) + .low_watermark(low_watermark) + .error_code(error_code.into()), + ); + } + } + + responses.push( + DeleteRecordsTopicResult::default() + .name(topic.name.clone()) + .partitions(Some(partition_responses)), + ); + } + + tx.commit().await.inspect_err(|err| error!(?err))?; + + Ok(responses) } async fn delete_topic(&self, topic: &TopicId) -> Result { @@ -2060,9 +2159,12 @@ impl Storage for Engine { .and_then(|offset| { row.get_value(1) .map_err(Into::into) - .and_then(LiteTimestamp::try_from) - .map(SystemTime::from) - .map(Some) + .and_then(|value| match value { + Value::Null => Ok(None), + value => LiteTimestamp::try_from(value) + .map(SystemTime::from) + .map(Some), + }) .map(|timestamp| { debug!( cluster = self.cluster, @@ -4353,4 +4455,304 @@ mod tests { Ok(()) } + + /// `#[ignore]`d like every other engine-level test here: turso/limbo + /// cannot yet run the full produce flow in CI. Run manually against + /// turso with `cargo nextest run --features turso -- --ignored`. + #[ignore] + #[tokio::test] + async fn delete_records_advances_log_start() -> Result<()> { + use nisshi_sans_io::delete_records_request::DeleteRecordsPartition; + + let _guard = init_tracing()?; + + let temp_dir = tempdir().inspect(|temporary| debug!(?temporary))?; + let file_path = temp_dir.path().join("nisshi.db"); + + let storage = Url::parse(&format!("file://{}", file_path.display()))?; + let cluster = "nisshi"; + let node = 12321; + + let engine = Engine::builder() + .advertised_listener(Url::parse("tcp://127.0.0.1:9092")?) + .cluster(cluster.to_owned()) + .storage(storage) + .node(node) + .build() + .await?; + + engine + .register_broker(BrokerRegistrationRequest { + broker_id: node, + cluster_id: cluster.to_owned(), + incarnation_id: Uuid::new_v4(), + rack: None, + }) + .await?; + + _ = engine + .create_topic( + CreatableTopic::default() + .name("t".into()) + .num_partitions(1) + .replication_factor(1) + .assignments(Some([].into())) + .configs(Some([].into())), + false, + ) + .await?; + + let topition = Topition::new("t", 0); + + let record_count = 6; + for n in 0..record_count { + let batch = inflated::Batch::builder() + .record(Record::builder().value(Some(Bytes::from(format!("record-{n}"))))) + .build() + .and_then(TryInto::try_into)?; + + _ = engine.produce(None, &topition, batch).await?; + } + + let cutoff = 3; + + let results = engine + .delete_records(&[DeleteRecordsTopic::default() + .name("t".into()) + .partitions(Some( + [DeleteRecordsPartition::default() + .partition_index(0) + .offset(cutoff)] + .into(), + ))]) + .await?; + + assert_eq!(1, results.len()); + let partitions = results[0].partitions.clone().unwrap_or_default(); + assert_eq!(1, partitions.len()); + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(partitions[0].error_code)? + ); + assert_eq!(cutoff, partitions[0].low_watermark); + + let stage = engine.offset_stage(&topition).await?; + assert_eq!(cutoff, stage.log_start); + assert_eq!(record_count, stage.high_watermark); + + Ok(()) + } + + /// `offset == -1` ("delete everything up to the high watermark") must + /// never remove the active batch holding `high_watermark - 1`: the log + /// start catches up to the high watermark, but the high watermark + /// itself, and hence `ListOffsets(Latest)`, is unaffected. + /// + /// `#[ignore]`d like every other engine-level test here: turso/limbo + /// cannot yet run the full produce flow in CI. Run manually against + /// turso with `cargo nextest run --features turso -- --ignored`. + #[ignore] + #[tokio::test] + async fn delete_records_to_high_watermark_keeps_latest() -> Result<()> { + use nisshi_sans_io::delete_records_request::DeleteRecordsPartition; + + let _guard = init_tracing()?; + + let temp_dir = tempdir().inspect(|temporary| debug!(?temporary))?; + let file_path = temp_dir.path().join("nisshi.db"); + + let storage = Url::parse(&format!("file://{}", file_path.display()))?; + let cluster = "nisshi"; + let node = 12321; + + let engine = Engine::builder() + .advertised_listener(Url::parse("tcp://127.0.0.1:9092")?) + .cluster(cluster.to_owned()) + .storage(storage) + .node(node) + .build() + .await?; + + engine + .register_broker(BrokerRegistrationRequest { + broker_id: node, + cluster_id: cluster.to_owned(), + incarnation_id: Uuid::new_v4(), + rack: None, + }) + .await?; + + _ = engine + .create_topic( + CreatableTopic::default() + .name("t".into()) + .num_partitions(1) + .replication_factor(1) + .assignments(Some([].into())) + .configs(Some([].into())), + false, + ) + .await?; + + let topition = Topition::new("t", 0); + + let record_count = 6; + for n in 0..record_count { + let batch = inflated::Batch::builder() + .record(Record::builder().value(Some(Bytes::from(format!("record-{n}"))))) + .build() + .and_then(TryInto::try_into)?; + + _ = engine.produce(None, &topition, batch).await?; + } + + let results = engine + .delete_records(&[DeleteRecordsTopic::default() + .name("t".into()) + .partitions(Some( + [DeleteRecordsPartition::default() + .partition_index(0) + .offset(-1)] + .into(), + ))]) + .await?; + + let partitions = results[0].partitions.clone().unwrap_or_default(); + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(partitions[0].error_code)? + ); + assert_eq!(record_count, partitions[0].low_watermark); + + let stage = engine.offset_stage(&topition).await?; + assert_eq!(record_count, stage.log_start); + assert_eq!( + record_count, stage.high_watermark, + "the active batch holding high_watermark - 1 must survive a -1 delete" + ); + + // `offset_stage` reads the watermark only, so these check the + // records: `Latest` needs the record at `high_watermark - 1`, and + // `Earliest` must answer the log start and not that record. + let offsets = engine + .list_offsets( + IsolationLevel::ReadUncommitted, + &[ + (topition.clone(), ListOffsetRequest::Latest), + (topition.clone(), ListOffsetRequest::Earliest), + ], + ) + .await?; + + assert_eq!(Some(record_count), offsets[0].1.offset, "{offsets:?}"); + assert_eq!(Some(record_count), offsets[1].1.offset, "{offsets:?}"); + + Ok(()) + } + + /// An unknown topic in a `DeleteRecords` request must never fail the + /// whole call: it reports `UnknownTopicOrPartition` for that one + /// partition while a sibling topic in the same request still succeeds. + /// + /// `#[ignore]`d like every other engine-level test here: turso/limbo + /// cannot yet run the full produce flow in CI. Run manually against + /// turso with `cargo nextest run --features turso -- --ignored`. + #[ignore] + #[tokio::test] + async fn delete_records_unknown_topic_does_not_fail_sibling() -> Result<()> { + use nisshi_sans_io::delete_records_request::DeleteRecordsPartition; + + let _guard = init_tracing()?; + + let temp_dir = tempdir().inspect(|temporary| debug!(?temporary))?; + let file_path = temp_dir.path().join("nisshi.db"); + + let storage = Url::parse(&format!("file://{}", file_path.display()))?; + let cluster = "nisshi"; + let node = 12321; + + let engine = Engine::builder() + .advertised_listener(Url::parse("tcp://127.0.0.1:9092")?) + .cluster(cluster.to_owned()) + .storage(storage) + .node(node) + .build() + .await?; + + engine + .register_broker(BrokerRegistrationRequest { + broker_id: node, + cluster_id: cluster.to_owned(), + incarnation_id: Uuid::new_v4(), + rack: None, + }) + .await?; + + _ = engine + .create_topic( + CreatableTopic::default() + .name("t".into()) + .num_partitions(1) + .replication_factor(1) + .assignments(Some([].into())) + .configs(Some([].into())), + false, + ) + .await?; + + let topition = Topition::new("t", 0); + + let batch = inflated::Batch::builder() + .record(Record::builder().value(Some(Bytes::from_static(b"one")))) + .build() + .and_then(TryInto::try_into)?; + + _ = engine.produce(None, &topition, batch).await?; + + let results = engine + .delete_records(&[ + DeleteRecordsTopic::default() + .name("missing".into()) + .partitions(Some( + [DeleteRecordsPartition::default() + .partition_index(0) + .offset(0)] + .into(), + )), + DeleteRecordsTopic::default() + .name("t".into()) + .partitions(Some( + [DeleteRecordsPartition::default() + .partition_index(0) + .offset(1)] + .into(), + )), + ]) + .await?; + + assert_eq!(2, results.len()); + + let missing = results + .iter() + .find(|topic| topic.name == "missing") + .expect("missing topic result"); + let missing_partitions = missing.partitions.clone().unwrap_or_default(); + assert_eq!( + ErrorCode::UnknownTopicOrPartition, + ErrorCode::try_from(missing_partitions[0].error_code)? + ); + + let known = results + .iter() + .find(|topic| topic.name == "t") + .expect("known topic result"); + let known_partitions = known.partitions.clone().unwrap_or_default(); + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(known_partitions[0].error_code)? + ); + assert_eq!(1, known_partitions[0].low_watermark); + + Ok(()) + } } diff --git a/nisshi-storage-sql/src/lite.rs b/nisshi-storage-sql/src/lite.rs index ca5ee2054..1dccbf79d 100644 --- a/nisshi-storage-sql/src/lite.rs +++ b/nisshi-storage-sql/src/lite.rs @@ -30,7 +30,7 @@ use nisshi_sans_io::{ create_topics_request::CreatableTopic, delete_groups_response::DeletableGroupResult, delete_records_request::DeleteRecordsTopic, - delete_records_response::DeleteRecordsTopicResult, + delete_records_response::{DeleteRecordsPartitionResult, DeleteRecordsTopicResult}, describe_cluster_response::DescribeClusterBroker, describe_configs_response::{DescribeConfigsResourceResult, DescribeConfigsResult}, describe_topic_partitions_response::{ @@ -51,12 +51,12 @@ use nisshi_schema::{ redact_url, }; use nisshi_storage::{ - ArcDynStorage, BrokerRegistrationRequest, ChannelRequestLayer, Error, GroupDetail, - ListOffsetResponse, METER, MetadataResponse, NamedGroupDetail, OffsetCommitRequest, - OffsetStage, ProducerIdResponse, RequestChannelService, RequestStorageService, Result, - ScramCredential, SemaphoreProxy, Storage, TopicId, Topition, TxnAddPartitionsRequest, - TxnAddPartitionsResponse, TxnOffsetCommitRequest, TxnState, UpdateError, Version, - bounded_channel, + ArcDynStorage, BrokerRegistrationRequest, ChannelRequestLayer, + DELETE_RECORDS_INVALID_LOW_WATERMARK, Error, GroupDetail, ListOffsetResponse, METER, + MetadataResponse, NamedGroupDetail, OffsetCommitRequest, OffsetStage, ProducerIdResponse, + RequestChannelService, RequestStorageService, Result, ScramCredential, SemaphoreProxy, Storage, + TopicId, Topition, TxnAddPartitionsRequest, TxnAddPartitionsResponse, TxnOffsetCommitRequest, + TxnState, UpdateError, Version, bounded_channel, delete_records_cutoff, }; use opentelemetry::{ KeyValue, @@ -2514,7 +2514,101 @@ impl Storage for Delegate { topics: &[DeleteRecordsTopic], ) -> Result> { debug!(?topics); - todo!() + + let pc = self.connection().await?; + let tx = pc.transaction().await?; + + let mut responses = vec![]; + + for topic in topics { + let mut partition_responses = vec![]; + + if let Some(ref partitions) = topic.partitions { + for partition in partitions { + let topition = Topition::new(topic.name.as_str(), partition.partition_index); + + let (error_code, low_watermark) = + match self.watermark_select_for_update(&topition, &pc).await { + Err(Error::Api(error_code)) => { + (error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK) + } + Err(err) => return Err(err), + + Ok((low, high)) => { + let stage = OffsetStage { + last_stable: high.unwrap_or_default(), + high_watermark: high.unwrap_or_default(), + log_start: low.unwrap_or_default(), + }; + + match delete_records_cutoff(partition.offset, &stage) { + Err(error_code) => { + (error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK) + } + + Ok(cutoff) => { + // This keeps the record at `high_watermark - 1`, because + // `list_latest_offset_committed.sql` and + // `list_latest_offset_uncommitted.sql` answer `Latest` from + // that record. The watermark still advances to `cutoff`. + let physical_cutoff = cutoff.min(stage.high_watermark - 1); + + if physical_cutoff > stage.log_start { + _ = pc + .execute( + "record_delete_by_offset.sql", + ( + self.cluster.as_str(), + topic.name.as_str(), + partition.partition_index, + physical_cutoff, + ), + ) + .await + .inspect_err(|err| { + error!(?err, ?topition, physical_cutoff) + })?; + } + + _ = pc + .execute( + "watermark_update.sql", + ( + self.cluster.as_str(), + topic.name.as_str(), + partition.partition_index, + cutoff, + high.unwrap_or_default(), + ), + ) + .await + .inspect_err(|err| error!(?err, ?topition))?; + + (ErrorCode::None, cutoff) + } + } + } + }; + + partition_responses.push( + DeleteRecordsPartitionResult::default() + .partition_index(partition.partition_index) + .low_watermark(low_watermark) + .error_code(error_code.into()), + ); + } + } + + responses.push( + DeleteRecordsTopicResult::default() + .name(topic.name.clone()) + .partitions(Some(partition_responses)), + ); + } + + pc.commit(tx).await.inspect_err(|err| error!(?err))?; + + Ok(responses) } async fn delete_topic(&self, topic: &TopicId) -> Result { @@ -3358,9 +3452,12 @@ impl Storage for Delegate { .and_then(|offset| { row.get_value(1) .map_err(Into::into) - .and_then(LiteTimestamp::try_from) - .map(SystemTime::from) - .map(Some) + .and_then(|value| match value { + Value::Null => Ok(None), + value => LiteTimestamp::try_from(value) + .map(SystemTime::from) + .map(Some), + }) .map(|timestamp| { debug!( cluster = self.cluster, diff --git a/nisshi-storage-sql/src/pg.rs b/nisshi-storage-sql/src/pg.rs index ebecac730..7a8267e24 100644 --- a/nisshi-storage-sql/src/pg.rs +++ b/nisshi-storage-sql/src/pg.rs @@ -62,10 +62,11 @@ use nisshi_schema::{ lake::{House, LakeHouse as _}, }; use nisshi_storage::{ - BrokerRegistrationRequest, Error, GroupDetail, ListOffsetResponse, METER, MetadataResponse, - NamedGroupDetail, OffsetCommitRequest, OffsetStage, ProducerIdResponse, Result, - ScramCredential, Storage, TopicId, Topition, TxnAddPartitionsRequest, TxnAddPartitionsResponse, - TxnOffsetCommitRequest, TxnState, UpdateError, Version, + BrokerRegistrationRequest, DELETE_RECORDS_INVALID_LOW_WATERMARK, Error, GroupDetail, + ListOffsetResponse, METER, MetadataResponse, NamedGroupDetail, OffsetCommitRequest, + OffsetStage, ProducerIdResponse, Result, ScramCredential, Storage, TopicId, Topition, + TxnAddPartitionsRequest, TxnAddPartitionsResponse, TxnOffsetCommitRequest, TxnState, + UpdateError, Version, delete_records_cutoff, }; use opentelemetry::metrics::Histogram; use opentelemetry::{KeyValue, metrics::Counter}; @@ -77,7 +78,7 @@ use tokio_postgres::{ error::SqlState, types::{BorrowToSql, ToSql, Type}, }; -use tracing::{debug, error, instrument}; +use tracing::{debug, error, instrument, warn}; use url::Url; use uuid::Uuid; @@ -432,6 +433,94 @@ impl Postgres { } } + /// Advances one partition's log start for DeleteRecords, answering the + /// error code and new low watermark. + /// + /// The watermark advance is the delete: it commits first, in a short + /// transaction that holds the watermark row lock. Removing the records + /// below it follows in a separate transaction under the default statement + /// timeout. That removal is best effort: a failure is logged, and + /// `maintain` removes the rows on its next run. + async fn delete_records_partition( + &self, + c: &mut Object, + topition: &Topition, + offset: i64, + ) -> Result<(ErrorCode, i64)> { + let tx = c.transaction().await.inspect_err(|err| error!(?err))?; + + let (low, high) = self.watermark_select_for_update(topition, &tx).await?; + + let stage = OffsetStage { + last_stable: high.unwrap_or_default(), + high_watermark: high.unwrap_or_default(), + log_start: low.unwrap_or_default(), + }; + + let cutoff = match delete_records_cutoff(offset, &stage) { + Ok(cutoff) => cutoff, + Err(error_code) => return Ok((error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK)), + }; + + _ = self + .tx_prepare_execute( + &tx, + "watermark_update.sql", + &[ + &self.cluster, + &topition.topic(), + &topition.partition(), + &cutoff, + &high.unwrap_or_default(), + ], + ) + .await + .inspect_err(|err| error!(?err))?; + + tx.commit().await.inspect_err(|err| error!(?err))?; + + // This keeps the record at `high_watermark - 1`, because + // `list_latest_offset_committed.sql` and + // `list_latest_offset_uncommitted.sql` answer `Latest` from that + // record. A produce only inserts at or above the high watermark, so + // these rows need no watermark lock. + let physical_cutoff = cutoff.min(stage.high_watermark - 1); + + if cutoff > stage.log_start + && let Err(err) = self.record_delete_below(c, topition, physical_cutoff).await + { + warn!(?err, cluster = ?self.cluster, ?topition, physical_cutoff); + } + + Ok((ErrorCode::None, cutoff)) + } + + async fn record_delete_below( + &self, + c: &mut Object, + topition: &Topition, + physical_cutoff: i64, + ) -> Result { + let tx = c.transaction().await?; + + let deleted = self + .tx_prepare_execute( + &tx, + "record_delete_by_offset.sql", + &[ + &self.cluster, + &topition.topic(), + &topition.partition(), + &physical_cutoff, + ], + ) + .await?; + + tx.commit().await?; + + Ok(deleted) + } + fn attributes_for_error( &self, nickname: &str, @@ -1729,6 +1818,25 @@ impl Postgres { tx.commit().await.map_err(Into::into).and(Ok(deleted)) } + /// Removes the records that DeleteRecords left below each partition's log + /// start, when its own removal failed or timed out. + #[instrument(skip(self), ret)] + async fn record_delete_below_log_start(&self) -> Result { + let mut c = self.connection().await?; + let tx = c.transaction().await?; + + // see policy_compact: exempt this sweep from the default statement_timeout. + tx.batch_execute("SET LOCAL statement_timeout = 0") + .await + .inspect_err(|err| error!(?err))?; + + let deleted = self + .tx_prepare_execute(&tx, "record_delete_below_log_start.sql", &[&self.cluster]) + .await?; + + tx.commit().await.map_err(Into::into).and(Ok(deleted)) + } + async fn topic_with_key<'a>(&self, topic: &'a str) -> Result<(&'a str, Option<&'a str>)> { if let Some((base, key)) = topic.split_once('/') && self @@ -1925,125 +2033,52 @@ impl Storage for Postgres { ) -> Result> { debug!(cluster = self.cluster, ?topics); - let c = self.connection().await?; - - let delete_records = c - .prepare(concat!( - "delete from record", - " using topic, cluster", - " where", - " cluster.name=$1", - " and topic.name = $2", - " and record.partition = $3", - " and record.id >= $4", - " and topic.cluster = cluster.id", - " and record.topic = topic.id", - )) - .await - .inspect_err(|err| error!(?err, ?topics))?; - - let mut responses = vec![]; + let mut c = self.connection().await?; + let mut responses = Vec::with_capacity(topics.len()); + // Each partition commits on its own, so the request holds at most one + // watermark row lock at a time. It cannot then deadlock with EndTxn, + // and a produce waits only while its own partition advances. for topic in topics { - let mut partition_responses = vec![]; + let mut partitions = Vec::with_capacity(topic.partitions.as_ref().map_or(0, Vec::len)); - if let Some(ref partitions) = topic.partitions { - for partition in partitions { - _ = c - .execute( - &delete_records, - &[ - &self.cluster, - &topic.name, - &partition.partition_index, - &partition.offset, - ], - ) - .await - .inspect_err(|err| { - let cluster = self.cluster.as_str(); - let topic = topic.name.as_str(); - let partition_index = partition.partition_index; - let offset = partition.offset; - - error!(?err, ?cluster, ?topic, ?partition_index, ?offset) - })?; - - let prepared = c - .prepare(concat!( - "select", - " id as offset", - " from", - " record", - " join (", - " select", - " coalesce(min(record.id), (select last_value from record_id_seq)) as offset", - " from record, topic, cluster", - " where", - " topic.cluster = cluster.id", - " and cluster.name = $1", - " and topic.name = $2", - " and record.partition = $3", - " and record.topic = topic.id) as minimum", - " on record.id = minimum.offset", - )) - .await - .inspect_err(|err| { - let cluster = self.cluster.as_str(); - let topic = topic.name.as_str(); - let partition_index = partition.partition_index; - let offset = partition.offset; - - error!(?err, ?cluster, ?topic, ?partition_index, ?offset) - })?; + for partition in topic.partitions.iter().flatten() { + let topition = Topition::new(topic.name.as_str(), partition.partition_index); - let partition_result = c - .query_opt( - &prepared, - &[&self.cluster, &topic.name, &partition.partition_index], - ) - .await - .inspect_err(|err| { - let cluster = self.cluster.as_str(); - let topic = topic.name.as_str(); - let partition_index = partition.partition_index; - let offset = partition.offset; + let (error_code, low_watermark) = self + .delete_records_partition(&mut c, &topition, partition.offset) + .await + .unwrap_or_else(|err| match err { + Error::Api(error_code) => { + (error_code, DELETE_RECORDS_INVALID_LOW_WATERMARK) + } - error!(?err, ?cluster, ?topic, ?partition_index, ?offset) - }) - .map_or( - Ok(DeleteRecordsPartitionResult::default() - .partition_index(partition.partition_index) - .low_watermark(0) - .error_code(ErrorCode::UnknownServerError.into())), - |row| { - row.map_or( - Ok(DeleteRecordsPartitionResult::default() - .partition_index(partition.partition_index) - .low_watermark(0) - .error_code(ErrorCode::UnknownServerError.into())), - |row| { - row.try_get::<_, i64>(0).map(|low_watermark| { - DeleteRecordsPartitionResult::default() - .partition_index(partition.partition_index) - .low_watermark(low_watermark) - .error_code(ErrorCode::None.into()) - }) - }, - ) - }, - )?; + // Retriable, as Kafka answers a storage failure. A retry + // is safe: at or below the log start it changes nothing. + err => { + error!(?err, cluster = ?self.cluster, ?topition); + ( + ErrorCode::KafkaStorageError, + DELETE_RECORDS_INVALID_LOW_WATERMARK, + ) + } + }); - partition_responses.push(partition_result); - } + partitions.push( + DeleteRecordsPartitionResult::default() + .partition_index(partition.partition_index) + .low_watermark(low_watermark) + .error_code(error_code.into()), + ); } responses.push( DeleteRecordsTopicResult::default() .name(topic.name.clone()) - .partitions(Some(partition_responses)), + .partitions(Some(partitions)), ); } + Ok(responses) } @@ -2757,7 +2792,7 @@ impl Storage for Postgres { debug!(?row); row.try_get::<_, i64>(0).map(Some).and_then(|offset| { - row.try_get::<_, SystemTime>(1).map(Some).map(|timestamp| { + row.try_get::<_, Option>(1).map(|timestamp| { debug!( cluster = self.cluster, ?topition, @@ -3941,6 +3976,12 @@ impl Storage for Postgres { let deleted = self.policy_delete(now).await?; debug!(deleted); + // Best effort: a failure here must not stop compaction or the lake. + match self.record_delete_below_log_start().await { + Ok(below_log_start) => debug!(below_log_start), + Err(err) => warn!(?err), + } + let compacted = self.policy_compact().await?; debug!(compacted); @@ -4307,6 +4348,112 @@ mod tests { _ = storage.policy_compact().await?; _ = storage.policy_delete(SystemTime::now()).await?; + _ = storage.record_delete_below_log_start().await?; + + Ok(()) + } + + /// DeleteRecords commits the new log start before it removes the records + /// below it. When that removal fails, `maintain` removes them, and keeps + /// the record at `high_watermark - 1` that `Latest` reads. + #[tokio::test] + async fn maintain_removes_records_left_below_log_start() -> Result<()> { + let cluster = alphanumeric_string(15); + + let storage = Postgres::builder(CONNECTION)? + .cluster(cluster.as_str()) + .node(rng().random_range(0..i32::MAX)) + .build(); + + if let Err(err) = storage.connection().await { + eprintln!("skipping maintain_removes_records_left_below_log_start: {err:?}"); + return Ok(()); + } + + storage + .register_broker(BrokerRegistrationRequest { + broker_id: 111, + cluster_id: cluster.clone(), + incarnation_id: Uuid::now_v7(), + rack: None, + }) + .await?; + + let topic_name = alphanumeric_string(15); + + _ = storage + .create_topic( + CreatableTopic::default() + .name(topic_name.clone()) + .num_partitions(1) + .replication_factor(0) + .assignments(Some([].into())) + .configs(Some([].into())), + false, + ) + .await?; + + let topition = Topition::new(topic_name.clone(), 0); + let record_count = 6i64; + + for _ in 0..record_count { + let batch = Batch::builder() + .record(Record::builder().value(Bytes::from_static(b"r").into())) + .build() + .and_then(TryInto::try_into)?; + + _ = storage.produce(None, &topition, batch).await?; + } + + let stored = async || -> Result { + let c = storage.connection().await?; + let row = c + .query_one( + "select count(*) from record r \ + join topition tp on r.topition = tp.id \ + join topic t on tp.topic = t.id \ + join cluster c on t.cluster = c.id \ + where c.name = $1 and t.name = $2", + &[&cluster, &topic_name], + ) + .await?; + row.try_get::<_, i64>(0).map_err(Into::into) + }; + + // the log start advances, as a DeleteRecords whose removal failed leaves it + let advance_log_start = async |low: i64| -> Result<()> { + let mut c = storage.connection().await?; + let tx = c.transaction().await?; + + _ = storage + .tx_prepare_execute( + &tx, + "watermark_update.sql", + &[&cluster, &topic_name, &0i32, &low, &record_count], + ) + .await?; + + tx.commit().await.map_err(Into::into) + }; + + advance_log_start(3).await?; + assert_eq!(record_count, stored().await?); + + storage.maintain(SystemTime::now()).await?; + assert_eq!(record_count - 3, stored().await?); + + advance_log_start(record_count).await?; + storage.maintain(SystemTime::now()).await?; + assert_eq!(1, stored().await?); + + let latest = storage + .list_offsets( + IsolationLevel::ReadUncommitted, + &[(topition.clone(), ListOffset::Latest)], + ) + .await?; + + assert_eq!(Some(record_count), latest[0].1.offset); Ok(()) } diff --git a/nisshi-storage-sql/src/sql.rs b/nisshi-storage-sql/src/sql.rs index eb20de06c..11b72d434 100644 --- a/nisshi-storage-sql/src/sql.rs +++ b/nisshi-storage-sql/src/sql.rs @@ -199,6 +199,14 @@ pub(crate) static SQL: LazyLock = LazyLock::new(|| { include_sql!("sql/producer_update_sequence.sql"), ), ("record_copy.sql", include_sql!("sql/record_copy.sql")), + ( + "record_delete_by_offset.sql", + include_sql!("sql/record_delete_by_offset.sql"), + ), + ( + "record_delete_below_log_start.sql", + include_sql!("sql/record_delete_below_log_start.sql"), + ), ( "record_delete_by_topic.sql", include_sql!("sql/record_delete_by_topic.sql"), diff --git a/nisshi-storage-sql/src/sql/list_earliest_offset.sql b/nisshi-storage-sql/src/sql/list_earliest_offset.sql index 8e4a06971..b74a11621 100644 --- a/nisshi-storage-sql/src/sql/list_earliest_offset.sql +++ b/nisshi-storage-sql/src/sql/list_earliest_offset.sql @@ -14,16 +14,38 @@ -- limitations under the License. -- prepare list_earliest_offset (text, text, integer) as -select -r.offset_id, -r.timestamp +-- Kafka answers Earliest with the log start offset, which is watermark.low. +-- DeleteRecords advances watermark.low and keeps the record at +-- watermark.high - 1, so a record below watermark.low can remain. The query +-- answers with the first record at or above watermark.low, or with +-- watermark.low and no timestamp when none remains. +-- +-- The first branch takes its own limit 1 so it stays one probe of the +-- record primary key. With only the outer limit, the constant o hides the +-- index order and the engine can read every record at or above +-- watermark.low before choosing one. + +select offset_id, timestamp + +from + +(select + +1 as o, offset_id, timestamp + +from + +(select + +r.offset_id, r.timestamp from cluster c join topic t on t.cluster = c.id join topition tp on tp.topic = t.id +left join watermark w on w.topition = tp.id join record r on r.topition = tp.id where @@ -31,6 +53,30 @@ where c.name = $1 and t.name = $2 and tp.partition = $3 +and r.offset_id >= coalesce(w.low, 0) order by r.offset_id asc -limit 1; +limit 1) first_record + +union all + +select + +2 as o, w.low, null + +from + +cluster c +join topic t on t.cluster = c.id +join topition tp on tp.topic = t.id +join watermark w on w.topition = tp.id + +where + +c.name = $1 +and t.name = $2 +and tp.partition = $3 +and w.low is not null + +order by o, offset_id asc +limit 1) earliest; diff --git a/nisshi-storage-sql/src/sql/record_delete_below_log_start.sql b/nisshi-storage-sql/src/sql/record_delete_below_log_start.sql new file mode 100644 index 000000000..4d0beaf31 --- /dev/null +++ b/nisshi-storage-sql/src/sql/record_delete_below_log_start.sql @@ -0,0 +1,36 @@ +-- -*- mode: sql; sql-product: postgres; -*- +-- Copyright ⓒ 2024-2026 Peter Morgan +-- +-- Licensed under the Apache License, Version 2.0 (the "License"); +-- you may not use this file except in compliance with the License. +-- You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +-- prepare record_delete_below_log_start (text) as + +-- Removes the records that DeleteRecords left below a partition's log start +-- (watermark.low). It keeps the record at watermark.high - 1, which the +-- Latest queries join. Postgres's least() ignores a null, so both +-- watermarks must be present. + +delete from record + +using cluster c, topic t, topition tp, watermark w + +where + +c.name = $1 +and t.cluster = c.id +and tp.topic = t.id +and w.topition = tp.id +and record.topition = tp.id +and w.low is not null +and w.high is not null +and record.offset_id < least(w.low, w.high - 1); diff --git a/nisshi-storage-sql/src/sql/record_delete_by_offset.sql b/nisshi-storage-sql/src/sql/record_delete_by_offset.sql new file mode 100644 index 000000000..b9e011a3a --- /dev/null +++ b/nisshi-storage-sql/src/sql/record_delete_by_offset.sql @@ -0,0 +1,30 @@ +-- -*- mode: sql; sql-product: postgres; -*- +-- Copyright ⓒ 2024-2026 Peter Morgan +-- +-- Licensed under the Apache License, Version 2.0 (the "License"); +-- you may not use this file except in compliance with the License. +-- You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +-- Used by DeleteRecords: removes every record strictly below the new low +-- watermark (the resolved cutoff offset) for one topic partition. The +-- cutoff itself must already be validated (see +-- `nisshi_storage::delete_records_cutoff`) before this runs. +delete from record +where record.topition in ( + select tp.id + from cluster c + join topic t on t.cluster = c.id + join topition tp on tp.topic = t.id + where c.name = $1 + and t.name = $2 + and tp.partition = $3 +) +and record.offset_id < $4; diff --git a/nisshi-storage/src/lib.rs b/nisshi-storage/src/lib.rs index eea46dc07..af1370003 100644 --- a/nisshi-storage/src/lib.rs +++ b/nisshi-storage/src/lib.rs @@ -843,6 +843,46 @@ impl OffsetStage { } } +/// The `low_watermark` of a `DeleteRecords` partition result that carries an +/// error, as Kafka's `DeleteRecordsResponse.INVALID_LOW_WATERMARK`. +pub const DELETE_RECORDS_INVALID_LOW_WATERMARK: i64 = -1; + +/// Validate a client-requested `DeleteRecords` offset against a partition's +/// current [`OffsetStage`], and resolve it to the new low watermark (log +/// start offset). +/// +/// Mirrors the validation Kafka applies in `Partition.deleteRecordsOnLeader` +/// / `UnifiedLog.maybeIncrementLogStartOffset`: +/// +/// - `requested == -1` means "delete everything up to the high watermark" +/// (what `kafka-delete-records.sh` sends by default): resolves to +/// `stage.high_watermark`. +/// - Any other negative offset, or one above the high watermark, is +/// [`ErrorCode::OffsetOutOfRange`]. +/// - An offset at or below the current log start is a no-op: it resolves to +/// the *existing* `stage.log_start`, not the requested value, so the log +/// start never moves backward. +/// - Otherwise the requested offset becomes the new log start. +/// +/// Every backend must call this before mutating any watermark or deleting +/// any record/object, so the five storage engines cannot drift from one +/// another on what counts as a valid request. +pub fn delete_records_cutoff(requested: i64, stage: &OffsetStage) -> Result { + if requested == -1 { + return Ok(stage.high_watermark); + } + + if requested < -1 || requested > stage.high_watermark { + return Err(ErrorCode::OffsetOutOfRange); + } + + if requested <= stage.log_start { + return Ok(stage.log_start); + } + + Ok(requested) +} + /// Group Member #[derive(Clone, Debug, Default, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)] pub struct GroupMember { @@ -2414,6 +2454,104 @@ mod tests { Ok(()) } + #[test] + fn delete_records_cutoff_minus_one_means_high_watermark() { + let stage = OffsetStage { + last_stable: 100, + high_watermark: 100, + log_start: 10, + }; + + assert_eq!(Ok(100), delete_records_cutoff(-1, &stage)); + } + + #[test] + fn delete_records_cutoff_rejects_other_negative_offsets() { + let stage = OffsetStage { + last_stable: 100, + high_watermark: 100, + log_start: 10, + }; + + assert_eq!( + Err(ErrorCode::OffsetOutOfRange), + delete_records_cutoff(-2, &stage) + ); + + assert_eq!( + Err(ErrorCode::OffsetOutOfRange), + delete_records_cutoff(i64::MIN, &stage) + ); + } + + #[test] + fn delete_records_cutoff_rejects_offset_above_high_watermark() { + let stage = OffsetStage { + last_stable: 100, + high_watermark: 100, + log_start: 10, + }; + + assert_eq!( + Err(ErrorCode::OffsetOutOfRange), + delete_records_cutoff(101, &stage) + ); + } + + #[test] + fn delete_records_cutoff_at_or_below_log_start_is_a_noop() { + let stage = OffsetStage { + last_stable: 100, + high_watermark: 100, + log_start: 10, + }; + + // Exactly at the current log start: unchanged. + assert_eq!(Ok(10), delete_records_cutoff(10, &stage)); + + // Below the current log start: the log start never moves backward. + assert_eq!(Ok(10), delete_records_cutoff(0, &stage)); + } + + #[test] + fn delete_records_cutoff_within_range_uses_requested_offset() { + let stage = OffsetStage { + last_stable: 100, + high_watermark: 100, + log_start: 10, + }; + + assert_eq!(Ok(50), delete_records_cutoff(50, &stage)); + } + + #[test] + fn delete_records_cutoff_allows_explicit_high_watermark() { + let stage = OffsetStage { + last_stable: 100, + high_watermark: 100, + log_start: 10, + }; + + assert_eq!(Ok(100), delete_records_cutoff(100, &stage)); + } + + #[test] + fn delete_records_cutoff_on_empty_partition() { + // A brand new partition: high_watermark == log_start == 0. + let stage = OffsetStage { + last_stable: 0, + high_watermark: 0, + log_start: 0, + }; + + assert_eq!(Ok(0), delete_records_cutoff(-1, &stage)); + assert_eq!(Ok(0), delete_records_cutoff(0, &stage)); + assert_eq!( + Err(ErrorCode::OffsetOutOfRange), + delete_records_cutoff(1, &stage) + ); + } + fn log_file_path() -> Result { thread::current() .name()