And use willow25 wrappers around those to permit nested relative encoding / decoding

@@ -0,0 +15,4 @@
+/// A trait for describing Willow stores compatible with the [`export_drop`] and ['import_drop`].
+///
+/// This trait requires the store is able to count the number of ['Entry`] included by an [`Area`].
+pub trait DropFormatStore: Store + PayloadPrefixStore + Clone {

@@ -0,0 +24,4 @@
+    ) -> impl Future<Output = Result<usize, Self::InternalError>>;
+}
+
+#[derive(Debug)]

@@ -0,0 +26,4 @@
+
+#[derive(Debug)]
+pub enum ExportDropError<ConsumerError, StoreErr> {
+    EmptyDrop,

@@ -0,0 +25,4 @@
+}
+
+#[derive(Debug)]
+pub enum ExportDropError<ConsumerError, StoreErr> {

@@ -0,0 +28,4 @@
+pub enum ExportDropError<ConsumerError, StoreErr> {
+    EmptyDrop,
+    StoreErr(StoreErr),
+    ConsumerProblem(ConsumerError),

@@ -0,0 +49,4 @@
+    // https://willowprotocol.org/specs/drop-format/index.html#willow_drop_format
+
+    // TODO: Re-order the areas in a way which would give the highest probability of an efficiently relatively encoded entries.
+    // TODO: De-duplicate entries from overlapping areas.

@@ -0,0 +69,4 @@
+        return Err(ExportDropError::EmptyDrop);
+    }
+
+    compact_u64::cu64_encode_standalone(entries_count as u64, consumer)

@@ -0,0 +6,4 @@
+    storage::{PayloadPrefixStore, Store},
+};
+use bab_rs::generic::storage::verifiable_streaming::SliceStreamingOptions;
+use ufotofu::{

@@ -0,0 +15,4 @@
+/// A trait for describing Willow stores compatible with the [`export_drop`] and ['import_drop`].
+///
+/// This trait requires the store is able to count the number of ['Entry`] included by an [`Area`].
+pub trait DropFormatStore: Store + PayloadPrefixStore + Clone {

@@ -0,0 +24,4 @@
+    ) -> impl Future<Output = Result<usize, Self::InternalError>>;
+}
+
+#[derive(Debug)]

@@ -0,0 +69,4 @@
+        return Err(ExportDropError::EmptyDrop);
+    }
+
+    compact_u64::cu64_encode_standalone(entries_count as u64, consumer)

@@ -0,0 +78,4 @@
+
+    // We can do this because these are the default values.
+    let mut prev_authorised = unsafe {
+        PossiblyAuthorisedEntry::new(entry_minus_one, auth_minus_one)

@@ -0,0 +110,4 @@
+                    .map_err(ExportDropError::ConsumerProblem)?;
+            }
+
+            compact_u64::write_tag(&mut header, 2, 0, u64::from(authorised.timestamp()));

@@ -0,0 +128,4 @@
+            }
+
+            consumer
+                .consume(ufotofu::prelude::Either::Left(header))

@@ -0,0 +154,4 @@
+                .await
+                .map_err(ExportDropError::ConsumerProblem)?;
+
+            compact_u64::cu64_encode(u64::from(authorised.timestamp()), 2, consumer)

@@ -0,0 +178,4 @@
+                .map_err(ExportDropError::ConsumerProblem)?;
+
+            if include_payloads && header & 0b0000_0001 == 0b0000_0001 {
+                // TODO: If the entry had its complete payload, consume it here.

@@ -0,0 +193,4 @@
+                        crate::storage::GetPayloadSliceError::ConsumerError(err) => ExportDropError::ConsumerProblem(err),
+                        crate::storage::GetPayloadSliceError::NoSuchEntry => panic!("The store did not have the payload for an entry this implementation was pretty sure it should have."),
+                    })?;
+            }

@@ -0,0 +202,4 @@
+    Ok(())
+}
+
+#[derive(Debug)]

@@ -0,0 +203,4 @@
+}
+
+#[derive(Debug)]
+pub enum ImportDropError<ProducerError, StoreError> {

@@ -0,0 +205,4 @@
+#[derive(Debug)]
+pub enum ImportDropError<ProducerError, StoreError> {
+    UnexpectedEndOfInput,
+    ProducerProblem(ProducerError),

@@ -0,0 +207,4 @@
+    UnexpectedEndOfInput,
+    ProducerProblem(ProducerError),
+    BadEntry,
+    Store(StoreError),

@@ -0,0 +184,4 @@
+                        authorised.namespace_id(),
+                        &authorised,
+                        None,
+                        consumer,

@@ -0,0 +230,4 @@
+/// If `ignore_nonzero_offset_partial_payload_slices` is `true`, then payload slices at non-zero offsets will be ignored, otherwise an error will be returned. As there is no known existing implementation for encoding drops which encodes payload slices at non-zero offsets, you probably want this to be `true`.
+pub async fn import_drop<Store, P>(
+    store: &mut Store,
+    producer: &mut P,

@@ -0,0 +246,4 @@
+
+    // We can do this because these are the default values.
+    let mut prev_authorised = unsafe {
+        PossiblyAuthorisedEntry::new(entry_minus_one, auth_minus_one)

@@ -0,0 +227,4 @@
+
+/// Decode [`Entry`]s (and possibly their corresponding payloads) from a [drop] and insert them into a [`Store`].
+///
+/// If `ignore_nonzero_offset_partial_payload_slices` is `true`, then payload slices at non-zero offsets will be ignored, otherwise an error will be returned. As there is no known existing implementation for encoding drops which encodes payload slices at non-zero offsets, you probably want this to be `true`.

@@ -0,0 +235,4 @@
+) -> Result<(), ImportDropError<P::Error, Store::InternalError>>
+where
+    Store: DropFormatStore,
+    P: BulkProducer<Item = u8, Final = ()>,

@@ -0,0 +339,4 @@
+                        .await
+                        .map_err(ImportDropError::Store)?;
+
+                    // If we have more than zero bytes, skip ahead by this much.

@@ -0,0 +504,4 @@
+        prev_authorised = decoded_authed_entry
+    }
+
+    todo!()

@@ -0,0 +368,4 @@
+                    )
+                    .await
+                    .map_err(|err| match err {
+                        crate::storage::AppendToPayloadPrefixError::ProducerError(err) => match err

@@ -0,0 +439,4 @@
+                    )
+                    .await
+                    .map_err(|err| match err {
+                        crate::storage::AppendToPayloadPrefixError::ProducerError(err) => match err

@@ -0,0 +400,4 @@
+                let mut limited = producer.to_limit(slice_length as usize);
+
+                if !entry_was_inserted {
+                    // We might already have some of the payload bytes!

@@ -0,0 +463,4 @@
+                    })?;
+            }
+            0b0000_0011 => {
+                // TODO: Our store traits don't yet have a way of ingesting partial slices at offsets.

@@ -0,0 +466,4 @@
+                // TODO: Our store traits don't yet have a way of ingesting partial slices at offsets.
+                // So if it was asked for, error here.
+                // Or just skip ahead to the next entry without ingesting any of the slices!
+                if ignore_nonzero_offset_partial_payload_slices {

@@ -193,0 +220,4 @@
+
+/// A trait for describing Willow stores which can count all the of entries in a given [`Area`].
+pub trait AreaCountStore: Store {
+    /// Return the count of [`Entry`] [included](https://willowprotocol.org/specs/grouping-entries/index.html#area_include) by a given [`Area`] within a given [namespace](https://willowprotocol.org/specs/data-model/index.html#namespace)

@@ -193,0 +225,4 @@
+        &self,
+        namespace_id: &NamespaceId,
+        area: &Area,
+    ) -> impl Future<Output = Result<usize, Self::InternalError>>;

@@ -0,0 +22,4 @@
+    StoreError(StoreError),
+    /// The consumer consuming the encoded bytes of the drop returned an error.
+    ConsumerError(ConsumerError),
+    /// The given consumer consumed a different number of bytes than was expected.

@@ -0,0 +105,4 @@
+            write_tag(&mut header, 2, 0, u64::from(authorised.timestamp()));
+            write_tag(&mut header, 2, 2, authorised.payload_length());
+
+            let mut bytes_available = 0;

@@ -0,0 +196,4 @@
+                if bytes_consumed != bytes_available {
+                    return Err(ExportDropError::ConsumedBytesMismatch);
+                }
+            }