Atomic storage API

[kevin-harrison]

Please make sure these boxes are checked, before submitting a new PR.

• You ran the local CI checker with ./check.sh with no errors
• You reference which issue is being closed in the PR text (if applicable)
• You updated the OmniPaxos book (if applicable)

Issues

Fix #97, Fix #49

Breaking Changes

• TheStorage trait has been reworked to facilitate atomic writes and allow for arbitrary crash-recovery. The indices passed to these functions now refer to the indices of the entry log as if no trimming or snapshotting has happened.
• PersistentStorage and its config structs have been reworked to use RocksDB and implement the new Storage trait. The dependence on commitlog is therefore gone, addressing issue Implement better trim functionality #49.
• All indices are now of type usize instead of u64. This includes the indices used in the Storage trait, Message structs, and Omnipaxos public functions.
• Both Storage::get_promise implementations in the omnipaxos_storage module now return None instead of the default Ballot when there is no promise ballot saved in storage.
• The batch_accept feature flag has been removed and is now the default behavior.
• The Promise, AcceptSync, and AcceptDecide message struct fields have been changed.

Other Changes

• MemoryStorage now implements the new Storage trait but its implementation of write_atomically does not guarantee atomicity (acceptable since memory can't persist state through crash-recovery).
• Batched entries are now correctly flushed when reconfiguring and when deciding entries.
• The syncing behavior in the the prepare phase in now consistent between both leaders and followers.

[haraldng]

Just some initial comments. I'll do a more detailed review after these comments are fixed.

For next time, please don't do refactorings before the actual feature is approved. E.g., here the changes for atomic storage wouldn't have been bloated with the changes for replacing u64 with usize (which is trivial).

[haraldng]

After our discussion yesterday, I looked at the current code to see how we can keep as much as possible of the current batching and at the same time introduce transaction for atomic storage. I think what we need is a struct InternalStorageTxn for InternalStorage. This is to enable the clean transaction object API in the Sequence Paxos code. When we handle the InternalStorageTxn in InternalStorage, we will know which entries to actually flush. And based on that we will create the real transaction for storage. So something like this:

// transaction object for InternalStorage
struct InternalStorageTxn {
    state_ops: Vec<StateOp>, // StateOp is an enum for modifying accepted_round, decided_idx, etc.
    append_op: AppendOp,       // AppendOp is an enum for append, append_on_prefix, etc.
    snapshot: Option<(usize, S)>  // snapshot index and snapshot
}

// handle_acceptsync() in SequencePaxos
let state_ops = vec![StorageOp::AcceptedRound(4), StorageOp::DecidedIdx(10)];
let append_op = AppendOp::AppendOnPrefix(accsync.sync_idx, accsync.suffix);
let internal_txn = InternalStorageTxn {state_ops, append_ops, ..};
self.internal_storage.do_txn(internal_txn);

// do_txn(it: InternalStorageTxn) in InternalStorage
let append_res = match it.append_op {
    Append(entries, do_batching) if do_batching => self.state_cache.append_entries(entries),
    Append(entries, do_batching) if !do_batching => self.state_cache.append_entries_without_batching(entries),
    // ... AppendOnPrefix etc.
};
let mut txn = StorageTxn::new();  // transaction object for Storage
txn.insert(it.state_ops);
if let Some(entries_to_be_flushed) = append_res {
    txn.insert(StorageOp::Append(entries_to_be_flushed));
}
if let Some((idx, snapshot)) = it.snapshot {
    txn.insert(StorageOp::Snapshot(idx, snapshot));
}
self.state_cache.real_log_len = self.storage.flush(txn)?;
Ok(self.get_accepted_idx())

[haraldng]

Good job! The design looks really good and seems to have made things a lot cleaner. The comments are mostly related to the code structure. Please fix them all or comment on places you have other ideas.

[haraldng]

There is nothing wrong with this, but it could be optimized.

If we in general only send Accepted after flushing then actually this is not strictly necessary. (any chosen entry must be flushed among a majority and if we haven't flushed it, then we will get it in the AcceptSync).

To prevent flushing stuff that will anyway get overwritten, we should perhaps do some checks here first. The most basic check is if the leader is more updated than us, we don't flush. There are probably more sophisticated checks we can do. For now I think we should do this basic check and leave other optimizations for the future.

[haraldng]

On a second thought, let's leave this as it is and we can create an issue and fix it in the future. So we don't add more complexity to this PR which already has many changes.

[haraldng]

Mostly just add more comments/docs. Can now go ahead and refactor the storage file.

[haraldng]

Overwrite instead of delete where possible in PersistentStorage.

[soruh]

All indices are now of type usize instead of u64. This includes the indices used in the Storage trait, Message structs, and Omnipaxos public functions.

I couldn't find the reason for this anywhere. It seems unfortunate to effectively constrain omnipaxos to 64-bit systems.
Using 32-bit ids and assuming 150_000 ops/s (taken from the EuroSys'23 paper quoted in the readme). Would give approximately 8 hours until the 32-bit id is exhausted...

Am I missing something obvious / what is the reason for this change?

[haraldng]

The change is not performance-related but rather for API reasons, we wanted reading from the omnipaxos log to be similar to the Rust Vec API.

Since it is now usize, the user should be able to use it with any compatible system.

[soruh]

That makes sense ergonomics wise, my concern would be that a system with a 32bit architecture and one with a 64bit architecture could not be part of the same cluster. Or, for that matter any 32 bit system could exhause the 32bit log index in reasonable time.