Skip to content

Commit eaa91ec

Browse files
joostjagerclaude
andcommitted
Add non-atomic ordered batch write method to KVStore traits
Adds `write_batch` method to both `KVStoreSync` and `KVStore` traits with default implementations that delegate to the single `write` method. - `BatchWriteEntry`: struct containing namespace, key, and data - `BatchWriteResult`: returns successful write count and optional error - Writes execute sequentially in order; stops on first error - Existing implementations automatically inherit the default behavior Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1 parent 6ed79d9 commit eaa91ec

1 file changed

Lines changed: 123 additions & 1 deletion

File tree

lightning/src/util/persist.rs

Lines changed: 123 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ use bitcoin::{BlockHash, Txid};
1919
use core::convert::Infallible;
2020
use core::future::Future;
2121
use core::ops::Deref;
22-
use core::pin::pin;
22+
use core::pin::{pin, Pin};
2323
use core::str::FromStr;
2424
use core::task;
2525

@@ -119,6 +119,54 @@ pub const OUTPUT_SWEEPER_PERSISTENCE_KEY: &str = "output_sweeper";
119119
/// updates applied to be current) with another implementation.
120120
pub const MONITOR_UPDATING_PERSISTER_PREPEND_SENTINEL: &[u8] = &[0xFF; 2];
121121

122+
/// An entry in a batch write operation.
123+
pub struct BatchWriteEntry {
124+
/// The primary namespace for this write.
125+
pub primary_namespace: String,
126+
/// The secondary namespace for this write.
127+
pub secondary_namespace: String,
128+
/// The key to write to.
129+
pub key: String,
130+
/// The data to write.
131+
pub buf: Vec<u8>,
132+
}
133+
134+
impl BatchWriteEntry {
135+
/// Creates a new batch write entry.
136+
pub fn new(
137+
primary_namespace: impl Into<String>, secondary_namespace: impl Into<String>,
138+
key: impl Into<String>, buf: Vec<u8>,
139+
) -> Self {
140+
Self {
141+
primary_namespace: primary_namespace.into(),
142+
secondary_namespace: secondary_namespace.into(),
143+
key: key.into(),
144+
buf,
145+
}
146+
}
147+
}
148+
149+
/// The result of a batch write operation.
150+
#[derive(Debug)]
151+
pub struct BatchWriteResult {
152+
/// The number of writes that completed successfully.
153+
pub successful_writes: usize,
154+
/// The error that occurred, if any. If `None`, all writes succeeded.
155+
pub error: Option<io::Error>,
156+
}
157+
158+
impl BatchWriteResult {
159+
/// Returns `true` if all writes succeeded.
160+
pub fn is_ok(&self) -> bool {
161+
self.error.is_none()
162+
}
163+
164+
/// Returns the error if one occurred, consuming the result.
165+
pub fn err(self) -> Option<io::Error> {
166+
self.error
167+
}
168+
}
169+
122170
/// Provides an interface that allows storage and retrieval of persisted values that are associated
123171
/// with given keys.
124172
///
@@ -191,6 +239,31 @@ pub trait KVStoreSync {
191239
fn list(
192240
&self, primary_namespace: &str, secondary_namespace: &str,
193241
) -> Result<Vec<String>, io::Error>;
242+
/// Persists multiple key-value pairs in a single batch operation.
243+
///
244+
/// Processes writes in order. Non-atomic: if an error occurs, earlier writes may have already
245+
/// been persisted and will not be rolled back. However, writes after the failed one are never
246+
/// started.
247+
///
248+
/// The default implementation iterates through entries and calls [`Self::write`] for each one.
249+
/// Implementations may override for optimized batch operations.
250+
fn write_batch(&self, entries: Vec<BatchWriteEntry>) -> BatchWriteResult {
251+
let mut successful_writes = 0;
252+
for entry in entries {
253+
match self.write(
254+
&entry.primary_namespace,
255+
&entry.secondary_namespace,
256+
&entry.key,
257+
entry.buf,
258+
) {
259+
Ok(()) => successful_writes += 1,
260+
Err(e) => {
261+
return BatchWriteResult { successful_writes, error: Some(e) };
262+
},
263+
}
264+
}
265+
BatchWriteResult { successful_writes, error: None }
266+
}
194267
}
195268

196269
/// A wrapper around a [`KVStoreSync`] that implements the [`KVStore`] trait. It is not necessary to use this type
@@ -246,6 +319,13 @@ where
246319

247320
async move { res }
248321
}
322+
323+
fn write_batch(
324+
&self, entries: Vec<BatchWriteEntry>,
325+
) -> impl Future<Output = BatchWriteResult> + 'static + MaybeSend {
326+
let res = self.0.write_batch(entries);
327+
async move { res }
328+
}
249329
}
250330

251331
/// Provides an interface that allows storage and retrieval of persisted values that are associated
@@ -343,6 +423,48 @@ pub trait KVStore {
343423
fn list(
344424
&self, primary_namespace: &str, secondary_namespace: &str,
345425
) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + MaybeSend;
426+
/// Persists multiple key-value pairs in a single batch operation.
427+
///
428+
/// Processes writes in order, awaiting each write before starting the next. Non-atomic: if an
429+
/// error occurs, earlier writes may have already been persisted and will not be rolled back.
430+
/// However, writes after the failed one are never started.
431+
///
432+
/// Note that similar to [`Self::write`], ordering is maintained: all writes in the batch are
433+
/// ordered relative to each other and to concurrent writes.
434+
///
435+
/// The default implementation calls [`Self::write`] for each entry sequentially.
436+
fn write_batch(
437+
&self, entries: Vec<BatchWriteEntry>,
438+
) -> impl Future<Output = BatchWriteResult> + 'static + MaybeSend {
439+
// Capture all write futures synchronously to maintain ordering
440+
// (version numbers are assigned when write() is called, not when awaited)
441+
let write_futures: Vec<Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send>>> =
442+
entries
443+
.into_iter()
444+
.map(|entry| {
445+
let fut = self.write(
446+
&entry.primary_namespace,
447+
&entry.secondary_namespace,
448+
&entry.key,
449+
entry.buf,
450+
);
451+
Box::pin(fut) as Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send>>
452+
})
453+
.collect();
454+
455+
async move {
456+
let mut successful_writes = 0;
457+
for write_future in write_futures {
458+
match write_future.await {
459+
Ok(()) => successful_writes += 1,
460+
Err(e) => {
461+
return BatchWriteResult { successful_writes, error: Some(e) };
462+
},
463+
}
464+
}
465+
BatchWriteResult { successful_writes, error: None }
466+
}
467+
}
346468
}
347469

348470
/// Provides additional interface methods that are required for [`KVStore`]-to-[`KVStore`]

0 commit comments

Comments
 (0)