feat: Parallellize snapshot creation

[segfault-magnet]

closes: #1789

Feature:
A SnapshotWriter can now be partial_closeed which results in a SnapshotFragment. Those can then be mergeed together and finalized into a SnapshotMetadata (and a written snapshot at that).

The export process now spawns rayon tasks for every table. At the end of the task it will partial_close its writer and return the SnapshotFragment.

When all tasks are finished the fragments are merged and the snapshot is done.

A TaskManager was written to deduplicate rayon task fail-fast and cancel logic.

Refactor:
Snapshot generation and import was moved together into fuel-core. Mainly because the TaskManager needed a place to live accesible to both. But also because if you're adding tables to the snapshot you'd also be wanting to read them so having both close together seems sensible.

[bvrooman]

It may be interesting to see if you can parameterize the DB query to work with a quantity and/or primary index, so that you can also use Rayon and the task manager for the write_contract_snapshot function.

[segfault-magnet]

Thought about it, but the contract_snapshot is there only to support writing e2e tests.

As I understood it you'd deploy yourself a contract, interact however you wish with it. When satisfied freeze that moment in time for e2e tests with the contract snapshot.

As such didn't feel a need to optimize for performance.

I'll give it a try anyway because the code might be more readable if I succeed.

[segfault-magnet]

Turned out nicely IMO. Have a look at lemme know your thoughts :)
Edit:
Also found a (probably) significant performance issue. Previously I had understood iter_all(prefix) to mean "iterate all entries whose key has this prefix.

But what it actually means is "start iterating thought the db at the first match of this prefix (if any). Then continue until the end of the db. That caused the contract snapshot logic to load all state after the match and then separate out what belongs to the contract in question. Fixed that, so noice :)

[bvrooman]

I think introducing the KeyFilter trait is a nice abstraction, as it allows us to build where-esque clauses for queries at the call site level. I'm surprised that this change to the DB interface was relatively low impact on the rest of the codebase, which is also nice.

As a side note, my understanding of iter_all(prefix) is that your initial understanding is actually correct. It is defined as :

    /// Returns an iterator over the all entries in the table with the specified prefix.
    fn iter_all_by_prefix<M, P>(
        &self,
        prefix: Option<P>,
    ) -> BoxedIter<super::Result<(M::OwnedKey, M::OwnedValue)>>
    where
        M: Mappable,
        P: AsRef<[u8]>,
        Self: IterableTable<M>,
    {
        self.iter_all_filtered::<M, P>(prefix, None, None)
    }

indicating that it should be filtering by prefix. If that's not the case, there might be something wrong at the implementation level.

[segfault-magnet]

I'm surprised that this change to the DB interface was relatively low impact

It's because even though the method sounds general, it was created only to support generating snapshots. Besides that nobody uses entries.

If that's not the case, there might be something wrong at the implementation level.

Interestingly enough that is not the case for rocksdb. If you look underneath you'll see the following (match on (prefix, start_key)):

            (Some(prefix), None) => {
                if direction == IterDirection::Reverse {
                    self.reverse_prefix_iter(prefix, column).into_boxed()
                } else {
                    // start iterating in a certain direction within the keyspace
                    let iter_mode = IteratorMode::From(
                        prefix,
                        convert_to_rocksdb_direction(direction),
                    );
                    let mut opts = ReadOptions::default();
                    opts.set_prefix_same_as_start(true);

                    self._iter_all(column, opts, iter_mode).into_boxed()
                }
            }

It looks like that the prefix is used just to start the iteration from where it matches, but it is not going to stop it. A simple test shows this happening:

 #[test]
    fn recreate_me() {
        let test = |mut db: Database<OnChain>| {
            let contract_id_1 = ContractId::from_str(
                "5962be5ebddc516cb4ed7d7e76365f59e0d231ac25b53f262119edf76564aab4",
            )
            .unwrap();

            let mut insert_empty_code = |id| {
                StorageMutate::<ContractsRawCode>::insert(&mut db, &id, &[]).unwrap()
            };
            insert_empty_code(contract_id_1);

            let contract_id_2 = ContractId::from_str(
                "5baf0dcae7c114f647f6e71f1723f59bcfc14ecb28071e74895d97b14873c5dc",
            )
            .unwrap();
            insert_empty_code(contract_id_2);

            use itertools::Itertools;
            let matched_keys: Vec<_> = db
                .iter_all_by_prefix::<ContractsRawCode, _>(Some(contract_id_1))
                .map_ok(|(k, _)| k)
                .try_collect()
                .unwrap();

            assert_eq!(matched_keys, vec![contract_id_1]);
        };

        let temp_dir = tempfile::tempdir().unwrap();
        let db = Database::<OnChain>::in_memory();
        // in memory passes
        test(db);

        let db = Database::<OnChain>::open_rocksdb(temp_dir.path(), 1024 * 1024 * 1024)
            .unwrap();
        // rocks db fails
        test(db);
    }

I'll raise this as a question internally and we can discuss there.