persist and recover shard states.

[raulk]

Closes #32.
Closes #24.
Closes #13.

This PR fleshes out another important part of the system: shard state persistence and resumption upon restart, and it also rewrites the mount registry/factory abstraction.

Persisting shard state

On every iteration of the event loop, we persist the current shard state. This is persisted in a datastore.Datastore provided by the user. We assume it to be namespaced.

Shards are persisted under their shard key, and the state is serialised as JSON. There is unused code for switching to CBOR in the future, if we find that to be more performant.

We can put on extra smarts to avoid persisting when unnecessary (e.g. when the persisted entity would be identical than the last one written). We can easily do this because PersistentShard is a comparable struct, so we could retain the last written PersistentShard in the shard state, calculate the new one, and skip the datastore put if they're identical. TODO in #44.

We also need to clean up the serialization logic. I don't like MarshalJSON and UnmarshalJSON doing this much.

Resuming the DAG store

On start, we restore the state from the Datastore. We iterate and recover all shards, overwriting some interim states:

• ShardStateServing is reset to ShardStateAvailable, as when we start, there is guaranteed to be 0 active consumers.
• ShardStateInitializing is reset to ShardStateNew, and we restart the shard initialization by dispatching tasks to the event loop.

Mount registry and factory

I revisited the mount.Registry and mount.MountFactory design. This has been simplified by removing the latter. Instead, we use a Mount instance as a template, and we clone it every time we instantiate a new mount of that type.

This allows us to set "environmental" properties that get applied automatically to all mounts of that type, as for example a JSON-RPC endpoint in the Lotus mount:

type LotusMount string {
    Endpoint string
}

var _ mount.Mount = (*LotusMount)(nil) // assume it implements the interface

r := mount.NewRegistry()
r.Register("lotus", &LotusMount{Endpoint: "ws://host:port/rpc/v0"})

m, _ = r.Instantiate(&url.URL{Scheme: "lotus", ...}) // m will have Endpoint set

An example of this is used in the tests with the new mount.FSMount, which I had to replace mount.BytesMount with for testing, since we can't afford to serialize full CAR files into base64 when encoding URLs every time that we persist shards.

Finally, the new registry design allows us to register the same mount type under many different schemes, with different templates. This is important if, for example, you have various fs.FS you want to serve from (for testing), or you want to cater to more than one Lotus deployment.

Since we do type matching, you do need to create new types when doing that:

type LotusMount1 struct { *LotusMount }
type LotusMount2 struct { *LotusMount }
type LotusMount3 struct { *LotusMount }

r := mount.NewRegistry()
r.Register("lotus1", &LotusMount1{&LotusMount{Endpoint: "ws://host1:port1/rpc/v0"}})
r.Register("lotus2", &LotusMount2{&LotusMount{Endpoint: "ws://host2:port2/rpc/v0"}})
r.Register("lotus3", &LotusMount3{&LotusMount{Endpoint: "ws://host3:port3/rpc/v0"}})

[aarshkshah1992]

Great stuff ! Just some nits.

• Needs some solid tests.
• I'm worried about the fsm Idempotency after resetting the state to ShardStateNew and ensuring we don't end up with more than one transient if we'd already fetched a copy before going down and restarting but I'll write a test for that.

[aarshkshah1992]

I think rather than keeping a track of this indexed field across restarts/or even at all, we can always ask the IndexRepo if it has the Index we need. That'll make it one less mutable state to update/track/keep in sync with the IndexRepo.

[raulk]

Yeah, that simplification is worth it. We can use StatFullIndex to check if the index exists.

[raulk]

#47

[aarshkshah1992]

nit: would be nice to have a comment here about which of these fields are persisted and which aren't.

[raulk]

Will add.