Skip to content

[ENH](wal3): pin fragment generations for conditional writes #7672

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rescrv merged 3 commits into from
Sep 2, 2026
+815 −194
Merged

[ENH](wal3): pin fragment generations for conditional writes #7672

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f97b811
[ENH](wal3): pin fragment generations for conditional writes
rescrv Sep 2, 2026
4508002
contention
rescrv Sep 2, 2026
9be2a53
Kimi's pass
rescrv Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 123 additions & 90 deletions rust/log-service/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,6 +481,8 @@ pub struct Metrics {
conditional_write_admission_rejections: opentelemetry::metrics::Counter<u64>,
/// The number of conditional push retries after a stale required fragment start.
conditional_write_required_start_retries: opentelemetry::metrics::Counter<u64>,
/// Time spent waiting to pin the next fragment generation.
conditional_write_fragment_pin_wait_us: opentelemetry::metrics::Histogram<u64>,
/// The number of successful conditional writes with an inserted offset.
conditional_write_success_with_offset: opentelemetry::metrics::Counter<u64>,
/// The number of successful conditional writes with durable contention and no offset.
Expand Down Expand Up @@ -528,6 +530,9 @@ impl Metrics {
conditional_write_required_start_retries: meter
.u64_counter("conditional_write_required_start_retries")
.build(),
conditional_write_fragment_pin_wait_us: meter
.u64_histogram("conditional_write_fragment_pin_wait_us")
.build(),
conditional_write_success_with_offset: meter
.u64_counter("conditional_write_success_with_offset")
.build(),
Expand Down Expand Up @@ -2553,100 +2558,119 @@ impl LogServer {
messages.push(buf);
}
let record_count = messages.len() as i32;
let first_inserted_record_offset =
if let Some(conditional_write) = conditional_write.as_ref() {
let admission_predicate = conditional_admission_predicate(conditional_write);
let mut append_result = None;
let mut validation_start =
LogPosition::from_offset(conditional_write.observed_log_offset);
let conditional_push_max_retries = self.config.conditional_push_retry_attempts();
for attempt in 0..conditional_push_max_retries {
let validated_tail = self
.validate_committed_log_for_conditional_write(
topology_name.as_ref(),
collection_id,
conditional_write,
validation_start,
)
.await?;
let append_options = AppendOptions::new(write_id_metadata.clone())
.with_admission_predicate(Arc::clone(&admission_predicate))
.with_required_fragment_start(validated_tail);
match log
.append_many_with_options(messages.clone(), Some(append_options))
.await
{
Ok(offset) => {
self.metrics
.conditional_write_success_with_offset
.add(1, &[]);
tracing::debug!(
%collection_id,
first_inserted_record_offset = offset.offset(),
"conditional write append succeeded"
);
append_result = Some(Some(offset));
break;
}
Err(wal3::Error::LogContentionDurable) => {
self.metrics
.conditional_write_success_without_offset
.add(1, &[]);
tracing::debug!(
%collection_id,
"conditional write append durably contended"
);
append_result = Some(None);
break;
}
Err(wal3::Error::AdmissionRejected) => {
self.metrics
.conditional_write_admission_rejections
.add(1, &[]);
self.metrics
.conditional_write_in_flight_conflicts
.add(1, &[]);
tracing::info!(
%collection_id,
"conditional write rejected by in-flight admission"
);
return Err(Status::aborted(CONDITIONAL_WRITE_CONFLICT_MESSAGE));
}
Err(wal3::Error::LogContentionRetry)
if attempt + 1 < conditional_push_max_retries =>
{
validation_start = validated_tail;
self.metrics
.conditional_write_required_start_retries
.add(1, &[]);
tracing::info!(
%collection_id,
attempt = attempt + 1,
"conditional write missed required start; retrying validation"
);
}
Err(err) => return Err(push_append_error_to_status(err)),
}
}
match append_result {
Some(append_result) => append_result,
None => {
return Err(Status::internal(
"conditional push retry loop exited without an append result",
));
}
}
} else {
let append_options = AppendOptions::new(write_id_metadata);
let message_bytes = messages.iter().map(Vec::len).sum::<usize>();
let first_inserted_record_offset = if let Some(conditional_write) =
conditional_write.as_ref()
{
let admission_predicate = conditional_admission_predicate(conditional_write);
let mut append_result = None;
let mut validation_start =
LogPosition::from_offset(conditional_write.observed_log_offset);
let conditional_push_max_retries = self.config.conditional_push_retry_attempts();
for attempt in 0..conditional_push_max_retries {
// Replicated logs serialize conditional writes in Spanner. S3-backed logs pin
// their in-memory fragment before validation so concurrent transactions can join
// the same publish.
let fragment_pin = if topology_name.is_none() {
let pin_started = Instant::now();
let pin_result = log
.acquire_fragment_pin(message_bytes)
.instrument(tracing::info_span!("acquire_conditional_fragment_pin"))
.await;
self.metrics.conditional_write_fragment_pin_wait_us.record(
u64::try_from(pin_started.elapsed().as_micros()).unwrap_or(u64::MAX),
&[],
);
Some(pin_result.map_err(push_append_error_to_status)?)
Comment thread
claude[bot] marked this conversation as resolved.
Outdated
} else {
None
};
let validated_tail = self
.validate_committed_log_for_conditional_write(
topology_name.as_ref(),
collection_id,
conditional_write,
validation_start,
)
.await?;
let append_options = AppendOptions::new(write_id_metadata.clone())
.with_admission_predicate(Arc::clone(&admission_predicate))
.with_required_fragment_start(validated_tail);
match log
.append_many_with_options(messages, Some(append_options))
.append_many_with_options(messages.clone(), Some(append_options), fragment_pin)
.await
{
Ok(offset) => Some(offset),
Err(wal3::Error::LogContentionDurable) => None,
Ok(offset) => {
self.metrics
.conditional_write_success_with_offset
.add(1, &[]);
tracing::debug!(
%collection_id,
first_inserted_record_offset = offset.offset(),
"conditional write append succeeded"
);
append_result = Some(Some(offset));
break;
}
Err(wal3::Error::LogContentionDurable) => {
self.metrics
.conditional_write_success_without_offset
.add(1, &[]);
tracing::debug!(
%collection_id,
"conditional write append durably contended"
);
append_result = Some(None);
break;
}
Err(wal3::Error::AdmissionRejected) => {
self.metrics
.conditional_write_admission_rejections
.add(1, &[]);
self.metrics
.conditional_write_in_flight_conflicts
.add(1, &[]);
tracing::info!(
%collection_id,
"conditional write rejected by in-flight admission"
);
return Err(Status::aborted(CONDITIONAL_WRITE_CONFLICT_MESSAGE));
}
Err(wal3::Error::LogContentionRetry)
if attempt + 1 < conditional_push_max_retries =>
{
validation_start = validated_tail;
self.metrics
.conditional_write_required_start_retries
.add(1, &[]);
tracing::info!(
%collection_id,
attempt = attempt + 1,
"conditional write missed required start; retrying validation"
);
}
Err(err) => return Err(push_append_error_to_status(err)),
}
};
}
match append_result {
Some(append_result) => append_result,
None => {
return Err(Status::internal(
"conditional push retry loop exited without an append result",
));
}
}
} else {
let append_options = AppendOptions::new(write_id_metadata);
match log
.append_many_with_options(messages, Some(append_options), None)
.await
{
Ok(offset) => Some(offset),
Err(wal3::Error::LogContentionDurable) => None,
Err(err) => return Err(push_append_error_to_status(err)),
}
};
let first_inserted_record_offset = first_inserted_record_offset
.map(|offset| {
offset.offset().try_into().map_err(|_| {
Expand Down Expand Up @@ -6753,18 +6777,27 @@ mod tests {

#[async_trait::async_trait]
impl LogWriterTrait for FakeLogWriter {
async fn acquire_fragment_pin(

@dbeglord dbeglord Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Head of line blocking. If one transaction is very slow, and holds the generation open, it'll block all future generations right? Or can multiple generations be open at once?

@rescrv rescrv Sep 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If one transaction is very slow you will get HOL on the generation. This is not a problem as there are a finite number of requests allowed in at once, and wal3 alredy does HOL blocking on the batch.

&self,
_reserved_bytes: usize,
) -> Result<wal3::FragmentPin, wal3::Error> {
unreachable!("fake log writer does not pin fragments")
}

async fn append_with_options(
&self,
message: Vec<u8>,
options: Option<AppendOptions>,
) -> Result<LogPosition, wal3::Error> {
self.append_many_with_options(vec![message], options).await
self.append_many_with_options(vec![message], options, None)
.await
}

async fn append_many_with_options(
&self,
_messages: Vec<Vec<u8>>,
options: Option<AppendOptions>,
_pin: Option<wal3::FragmentPin>,
) -> Result<LogPosition, wal3::Error> {
self.observed_options.lock().push(options);
let mut append_results = self.append_results.lock();
Expand Down
Loading
Loading