pub(super) fn new(database: &Arc<DatabaseData<FileManager>>) -> Self {

        Self {

            db: Arc::downgrade(database),

            scratch: Vec::new(),

        }

    }

        if let Some(database) = self.db.upgrade() {

            let mut written_grains = Vec::new();

            let mut freed_grains = Vec::new();

            let mut index_metadata = database.atlas.current_index_metadata()?;

                match entry.read_chunk()? {

                    ReadChunkResult::Chunk(mut chunk) => {

                        let position = chunk.log_position();

                        self.scratch

                            .resize(u32_to_usize(chunk.bytes_remaining())?, 0);

                        chunk.read_exact(&mut self.scratch)?;

                        let chunk = WalChunk::read(&self.scratch)?;

                        match chunk {

                            WalChunk::NewGrainInWal { id, .. } => {

                                written_grains.push((id, position))

                            WalChunk::FinishTransaction { commit_log_entry } => {

                                index_metadata.commit_log_head = Some(commit_log_entry);

                            }

                            WalChunk::UpdatedEmbeddedHeader(header) => {

                                index_metadata.embedded_header_data = header;

                            }

                    ReadChunkResult::EndOfEntry => break,

            freed_grains.sort_unstable();

            database.atlas.note_transaction_committed(

                index_metadata,

                written_grains,

                &freed_grains,

                true,

            )?;

        Ok(())

    }

        if let Some(database) = self.db.upgrade() {

            let database = Database {

                data: database,

                wal: wal.clone(),

            };

            let fsyncs = database.data.store.file_manager.new_fsync_batch()?;

            let latest_tx_id = TransactionId::from(last_checkpointed_id);

            let mut store = database.data.store.lock()?;

            let mut needs_directory_sync = store.needs_directory_sync;

            let mut all_changed_grains = Vec::new();

            let mut latest_commit_log_entry = store.index.active.commit_log_head;

            let mut latest_embedded_header_data = store.index.active.embedded_header_data;

            let mut latest_checkpoint_target = store.index.active.checkpoint_target;

            let mut latest_checkpointed_to = store.index.active.checkpointed_to;

            // We allocate the transaction grains vec once and reuse the vec to

            // avoid reallocating.

            let mut transaction_grains = Vec::new();

            'entry_loop: while let Some(mut entry) = checkpointed_entries.read_entry()? {

                let checkpointed_tx = TransactionId::from(entry.id());

                // Because an entry could be aborted, we need to make sure we don't

                // modify our DiskState until after we've read every chunk. We will

                // write new grain data directly to the segments, but the headers

                // won't be updated until after the loop as well.

                transaction_grains.clear();

                while let Some(mut chunk) = match entry.read_chunk()? {

                    ReadChunkResult::Chunk(chunk) => Some(chunk),

                    ReadChunkResult::EndOfEntry => None,

                    self.scratch.clear();

                    chunk.read_to_end(&mut self.scratch)?;

                    if !chunk.check_crc()? {

                    }

                    match WalChunk::read(&self.scratch)? {

                        WalChunk::NewGrainInWal { id, data } => {

                            let basin = store.basins.get_or_insert_with(id.basin_id(), || {

                                BasinState::default_for(id.basin_id())

                            });

                            let stratum = basin.get_or_allocate_stratum(

                                id.stratum_id(),

                                &database.data.store.directory,

                            );

                            let mut file = BufWriter::new(stratum.get_or_open_file(

                                &database.data.store.file_manager,

                                &mut needs_directory_sync,

                            )?);

                            let file_position = id.file_position();

                            file.seek(SeekFrom::Start(file_position))?;

                            file.write_all(&checkpointed_tx.to_be_bytes())?;

                            file.write_all(&usize_to_u32(data.len())?.to_be_bytes())?;

                            file.write_all(data)?;

                            let crc32 = crc32c::crc32c(data);

                            file.write_all(&crc32.to_be_bytes())?;

                            file.flush()?;

                            transaction_grains.push((id, GrainAllocationStatus::Allocated));

                        WalChunk::ArchiveGrain(id) => {

                            transaction_grains.push((id, GrainAllocationStatus::Archived));

                        }

                        WalChunk::FreeGrain(id) => {

                            transaction_grains.push((id, GrainAllocationStatus::Free));

                        }

                        WalChunk::FinishTransaction { commit_log_entry } => {

                            latest_commit_log_entry = Some(commit_log_entry);

                        }

                        WalChunk::UpdatedEmbeddedHeader(header) => {

                            latest_embedded_header_data = header;

                        }

                        WalChunk::CheckpointTo(tx_id) => {

                            latest_checkpoint_target = tx_id;

                        }

                        WalChunk::CheckpointedTo(tx_id) => {

                            latest_checkpointed_to = tx_id;

                        }

                all_changed_grains.append(&mut transaction_grains);

            all_changed_grains.sort_unstable();

            let mut index = 0;

            while let Some((first_id, _)) = all_changed_grains.get(index).cloned() {

                let basin = store.basins.get_or_insert_with(first_id.basin_id(), || {

                });

                let stratum = basin

                    .get_or_allocate_stratum(first_id.stratum_id(), &database.data.store.directory);

                    match all_changed_grains.get(index).copied() {

                        Some((id, status))

                            if id.basin_id() == first_id.basin_id()

                                && id.stratum_id() == first_id.stratum_id() =>

                        {

                            let local_index = usize::from(id.local_grain_index().as_u16());

                            if status == GrainAllocationStatus::Free {

                                // Free grains are just 0s.

                                stratum.header.active.grains

                                    [local_index..local_index + usize::from(id.grain_count())]

                                    .fill(0);

                            } else {

                                for index in 0..id.grain_count() {

                                    let status = if status == GrainAllocationStatus::Allocated {

                                        GrainAllocationInfo::allocated(id.grain_count() - index)

                                        GrainAllocationInfo::archived(id.grain_count() - index)

                                    stratum.header.active.grains

                                        [local_index + usize::from(index)] = status.0;

                        _ => break,

                    index += 1;

                stratum.write_header(latest_tx_id, &fsyncs)?;

            store.index.active.commit_log_head = latest_commit_log_entry;

            store.index.active.embedded_header_data = latest_embedded_header_data;

            store.index.active.checkpoint_target = latest_checkpoint_target;

            store.index.active.checkpointed_to = latest_checkpointed_to;

            store.write_header(latest_tx_id, &fsyncs)?;

            if needs_directory_sync {

                store.needs_directory_sync = false;

                fsyncs.queue_fsync_all(store.directory.try_clone()?)?;

            }

            fsyncs.wait_all()?;

            database

                .data

                .atlas

                .note_grains_checkpointed(&all_changed_grains)?;

            database

                .data

                .checkpointer

                .checkpoint_to(latest_checkpoint_target);

            Ok(())

    }

    pub fn read(buffer: &'a [u8]) -> Result<Self> {

        if buffer.len() < Self::COMMAND_LENGTH_USIZE {

            return Err(Error::ValueOutOfBounds);

        }

        let kind = buffer[0];

        match kind {

                id: GrainId::from_bytes(&buffer[1..9]).ok_or(Error::InvalidGrainId)?,

                data: &buffer[9..],

                GrainId::from_bytes(&buffer[1..9]).ok_or(Error::InvalidGrainId)?,

                GrainId::from_bytes(&buffer[1..9]).ok_or(Error::InvalidGrainId)?,

            3 => Ok(Self::UpdatedEmbeddedHeader(GrainId::from_bytes(

                &buffer[1..9],

            ))),

            4 => Ok(Self::CheckpointTo(TransactionId::from(EntryId(

                u64::from_be_bytes(buffer[1..9].try_into().expect("u64 is 8 bytes")),

            )))),

            5 => Ok(Self::CheckpointedTo(TransactionId::from(EntryId(

                u64::from_be_bytes(buffer[1..9].try_into().expect("u64 is 8 bytes")),

            )))),

                commit_log_entry: GrainId::from_bytes(&buffer[1..9])

                    .ok_or(Error::InvalidGrainId)?,

            _ => Err(Error::verification_failed("invalid wal chunk")),

    }

        writer.write_all(&[0])?;

        writer.write_all(&grain_id.to_be_bytes())?;

        writer.write_all(data)?;

        Ok(())

    }

        writer.write_all(&[1])?;

        writer.write_all(&grain_id.to_be_bytes())?;

        Ok(())

    }

        writer.write_all(&[2])?;

        writer.write_all(&grain_id.to_be_bytes())?;

        Ok(())

    }

        writer.write_all(&[3])?;

        writer.write_all(&new_embedded_header.unwrap_or(GrainId::NONE).to_be_bytes())?;

        Ok(())

    }

        writer.write_all(&[4])?;

        writer.write_all(&checkpoint_to.to_be_bytes())?;

        Ok(())

    }

        writer.write_all(&[5])?;

        writer.write_all(&checkpointed_to.to_be_bytes())?;

        Ok(())

    }

    pub const fn new_grain_length(data_length: u32) -> u32 {

        data_length + 9

    }

        writer.write_all(&[255])?;

        writer.write_all(&commit_log_entry_id.to_be_bytes())?;

        Ok(())

    }

#[test]

fn wal_chunk_error_tests() {

    // The valid WalChunks are all tested by virtue of testing sediment. The

    // errors, however, are nearly impossible to simulate due to the wal being

    // completely abstracted away.

    let Error::VerificationFailed(_) = WalChunk::read(&[254,0,0,0,0,0,0,0,0]).unwrap_err() else { unreachable!() };

    let Error::ValueOutOfBounds = WalChunk::read(&[254]).unwrap_err() else { unreachable!() };

}