feat: Add lens migration engine to defra (#1564)

## Relevant issue(s)

Resolves #1448 #1556

Adds lens as a migration engine to defra.

Migrations are run lazily, on fetch. This includes on updates (locally,
and via P2P). The migration result is cached within the datastore. The
DAG is never updated to reflect the migration result, including when the
migrations are executed during an update. The current schema version of
items in the datastore is tracked at the document level.

Commits should be clean, and contain some hopefully handy documentation,
although the bulk of the work is in the last, large commit.

.gitignore

@@ -7,6 +7,16 @@ coverage.txt
 tests/bench/*.log
 tests/bench/*.svg
 
+tests/lenses/rust_wasm32_set_default/Cargo.lock
+tests/lenses/rust_wasm32_set_default/target
+tests/lenses/rust_wasm32_set_default/pkg
+tests/lenses/rust_wasm32_remove/Cargo.lock
+tests/lenses/rust_wasm32_remove/target
+tests/lenses/rust_wasm32_remove/pkg
+tests/lenses/rust_wasm32_copy/Cargo.lock
+tests/lenses/rust_wasm32_copy/target
+tests/lenses/rust_wasm32_copy/pkg
+
 # Ignore OS X metadata files.
 .history
 **.DS_Store

CONTRIBUTING.md

@@ -55,6 +55,13 @@ Run the following commands for testing:
 - `make bench` to run the benchmark suite. To compare a branch's results with the `develop` branch results, execute the suite on both branches, output the results to files, and compare them with a tool like `benchstat` (e.g., `benchstat develop.txt current.txt`). To install `benchstat`, use `make deps:bench`.
 - `make test:changes` to run a test suite detecting breaking changes. Accompany breaking changes with documentation in `docs/data_format_changes/` for the test to pass.
 
+### Test prerequisites
+
+The following tools are required in order to build and run the tests within this repository:
+
+- [Go](https://go.dev/doc/install)
+- Cargo/rustc, typically installed via [rustup](https://www.rust-lang.org/tools/install)
+
 ## Documentation
 The overall project documentation can be found at [docs.source.network](https://docs.source.network), and its source at [github.com/sourcenetwork/docs.source.network](https://github.com/sourcenetwork/docs.source.network).
 

Makefile

@@ -27,6 +27,9 @@ endif
 
 TEST_FLAGS=-race -shuffle=on -timeout 70s
 
+LENS_TEST_DIRECTORY=tests/integration/schema/migrations
+DEFAULT_TEST_DIRECTORIES=$$(go list ./... | grep -v $(LENS_TEST_DIRECTORY))
+
 default:
 	@go run $(BUILD_FLAGS) cmd/defradb/main.go
 
@@ -73,9 +76,15 @@ deps\:lint:
 deps\:test:
 	go install gotest.tools/gotestsum@latest
 
+.PHONY: deps\:lens
+deps\:lens:
+	rustup target add wasm32-unknown-unknown
+	@$(MAKE) -C ./tests/lenses build
+
 .PHONY: deps\:coverage
 deps\:coverage:
 	go install github.com/ory/go-acc@latest
+	@$(MAKE) deps:lens
 
 .PHONY: deps\:bench
 deps\:bench:
@@ -114,6 +123,8 @@ mock:
 	mockery --dir ./datastore --output ./datastore/mocks --name RootStore --with-expecter
 	mockery --dir ./datastore --output ./datastore/mocks --name Txn --with-expecter
 	mockery --dir ./datastore --output ./datastore/mocks --name DAGStore --with-expecter
+	mockery --dir ./db/fetcher --output ./db/fetcher/mocks --name Fetcher --with-expecter
+	mockery --dir ./db/fetcher --output ./db/fetcher/mocks --name EncodedDocument --with-expecter
 
 .PHONY: dev\:start
 dev\:start:
@@ -162,32 +173,37 @@ endif
 
 .PHONY: test
 test:
-	gotestsum --format pkgname -- ./... $(TEST_FLAGS)
+	gotestsum --format pkgname -- $(DEFAULT_TEST_DIRECTORIES) $(TEST_FLAGS)
 
 # Only build the tests (don't execute them).
 .PHONY: test\:build
 test\:build:
-	gotestsum --format pkgname -- ./... $(TEST_FLAGS) -run=nope
+	gotestsum --format pkgname -- $(DEFAULT_TEST_DIRECTORIES) $(TEST_FLAGS) -run=nope
 
 .PHONY: test\:ci
 test\:ci:
-	DEFRA_BADGER_MEMORY=true DEFRA_BADGER_FILE=true $(MAKE) test:names
+	DEFRA_BADGER_MEMORY=true DEFRA_BADGER_FILE=true $(MAKE) test:all
 
 .PHONY: test\:go
 test\:go:
-	go test ./... $(TEST_FLAGS)
+	go test $(DEFAULT_TEST_DIRECTORIES) $(TEST_FLAGS)
 
 .PHONY: test\:names
 test\:names:
-	gotestsum --format testname -- ./... $(TEST_FLAGS)
+	gotestsum --format testname -- $(DEFAULT_TEST_DIRECTORIES) $(TEST_FLAGS)
+
+.PHONY: test\:all
+test\:all:
+	@$(MAKE) test:names
+	@$(MAKE) test:lens
 
 .PHONY: test\:verbose
 test\:verbose:
-	gotestsum --format standard-verbose -- ./... $(TEST_FLAGS)
+	gotestsum --format standard-verbose -- $(DEFAULT_TEST_DIRECTORIES) $(TEST_FLAGS)
 
 .PHONY: test\:watch
 test\:watch:
-	gotestsum --watch -- ./...
+	gotestsum --watch -- $(DEFAULT_TEST_DIRECTORIES)
 
 .PHONY: test\:clean
 test\:clean:
@@ -205,6 +221,11 @@ test\:bench-short:
 test\:scripts:
 	@$(MAKE) -C ./tools/scripts/ test
 
+.PHONY: test\:lens
+test\:lens:
+	@$(MAKE) deps:lens
+	gotestsum --format testname -- ./$(LENS_TEST_DIRECTORY)/... $(TEST_FLAGS)
+
 # Using go-acc to ensure integration tests are included.
 # Usage: `make test:coverage` or `make test:coverage path="{pathToPackage}"`
 # Example: `make test:coverage path="./api/..."`
@@ -231,6 +252,7 @@ test\:coverage-html:
 
 .PHONY: test\:changes
 test\:changes:
+	@$(MAKE) deps:lens
 	env DEFRA_DETECT_DATABASE_CHANGES=true gotestsum -- ./... -shuffle=on -p 1
 
 .PHONY: validate\:codecov

client/db.go

@@ -108,6 +108,25 @@ type Store interface {
 	// [FieldKindStringToEnumMapping].
 	PatchSchema(context.Context, string) error
 
+	// SetMigration sets the migration for the given source-destination schema version IDs. Is equivilent to
+	// calling `LensRegistry().SetMigration(ctx, cfg)`.
+	//
+	// There may only be one migration per schema version id.  If another migration was registered it will be
+	// overwritten by this migration.
+	//
+	// Neither of the schema version IDs specified in the configuration need to exist at the time of calling.
+	// This is to allow the migration of documents of schema versions unknown to the local node recieved by the
+	// P2P system.
+	//
+	// Migrations will only run if there is a complete path from the document schema version to the latest local
+	// schema version.
+	SetMigration(context.Context, LensConfig) error
+
+	// LensRegistry returns the LensRegistry in use by this database instance.
+	//
+	// It exposes several useful thread-safe migration related functions.
+	LensRegistry() LensRegistry
+
 	// GetCollectionByName attempts to retrieve a collection matching the given name.
 	//
 	// If no matching collection is found an error will be returned.

client/document.go

@@ -51,11 +51,17 @@ import (
 // @body: A document interface can be implemented by both a TypedDocument and a
 // UnTypedDocument, which use a schema and schemaless approach respectively.
 type Document struct {
-	key    DocKey
-	fields map[string]Field
-	values map[Field]Value
-	head   cid.Cid
-	mu     sync.RWMutex
+	key DocKey
+	// SchemaVersionID holds the id of the schema version that this document is
+	// currently at.
+	//
+	// Migrating the document will update this value to the output version of the
+	// migration.
+	SchemaVersionID string
+	fields          map[string]Field
+	values          map[Field]Value
+	head            cid.Cid
+	mu              sync.RWMutex
 	// marks if document has unsaved changes
 	isDirty bool
 }

client/lens.go

@@ -0,0 +1,83 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package client
+
+import (
+	"context"
+
+	"github.com/lens-vm/lens/host-go/config/model"
+	"github.com/sourcenetwork/immutable/enumerable"
+
+	"github.com/sourcenetwork/defradb/datastore"
+)
+
+// LensConfig represents the configuration of a Lens migration in Defra.
+type LensConfig struct {
+	// SourceSchemaVersionID is the ID of the schema version from which to migrate
+	// from.
+	//
+	// The source and destination versions must be next to each other in the history.
+	SourceSchemaVersionID string
+
+	// DestinationSchemaVersionID is the ID of the schema version from which to migrate
+	// to.
+	//
+	// The source and destination versions must be next to each other in the history.
+	DestinationSchemaVersionID string
+
+	// The configuration of the Lens module.
+	//
+	// For now, the wasm module must remain at the location specified as long as the
+	// migration is active.
+	model.Lens
+}
+
+// LensRegistry exposes several useful thread-safe migration related functions which may
+// be used to manage migrations.
+type LensRegistry interface {
+	// SetMigration sets the migration for the given source-destination schema version IDs. Is equivilent to
+	// calling `Store.SetMigration(ctx, cfg)`.
+	//
+	// There may only be one migration per schema version id.  If another migration was registered it will be
+	// overwritten by this migration.
+	//
+	// Neither of the schema version IDs specified in the configuration need to exist at the time of calling.
+	// This is to allow the migration of documents of schema versions unknown to the local node recieved by the
+	// P2P system.
+	//
+	// Migrations will only run if there is a complete path from the document schema version to the latest local
+	// schema version.
+	SetMigration(context.Context, datastore.Txn, LensConfig) error
+
+	// ReloadLenses clears any cached migrations, loads their configurations from the database and re-initializes
+	// them.  It is run on database start if the database already existed.
+	ReloadLenses(ctx context.Context, txn datastore.Txn) error
+
+	// MigrateUp returns an enumerable that feeds the given source through the Lens migration for the given
+	// schema version id if one is found, if there is no matching migration the given source will be returned.
+	MigrateUp(enumerable.Enumerable[map[string]any], string) (enumerable.Enumerable[map[string]any], error)
+
+	// MigrateDown returns an enumerable that feeds the given source through the Lens migration for the schema
+	// version that precedes the given schema version id in reverse, if one is found, if there is no matching
+	// migration the given source will be returned.
+	//
+	// This downgrades any documents in the source enumerable if/when enumerated.
+	MigrateDown(enumerable.Enumerable[map[string]any], string) (enumerable.Enumerable[map[string]any], error)
+
+	// Config returns a slice of the configurations of the currently loaded migrations.
+	//
+	// Modifying the slice does not affect the loaded configurations.
+	Config() []LensConfig
+
+	// HasMigration returns true if there is a migration registered for the given schema version id, otherwise
+	// will return false.
+	HasMigration(string) bool
+}

core/crdt/composite.go

@@ -154,13 +154,27 @@ func (c CompositeDAG) Merge(ctx context.Context, delta core.Delta, id string) er
 		return c.deleteWithPrefix(ctx, c.key.WithValueFlag().WithFieldId(""))
 	}
 
-	// ensure object marker exists
-	exists, err := c.store.Has(ctx, c.key.ToPrimaryDataStoreKey().ToDS())
+	// We cannot rely on the dagDelta.Status here as it may have been deleted locally, this is not
+	// reflected in `dagDelta.Status` if sourced via P2P.  Updates synced via P2P should not undelete
+	// the local reperesentation of the document.
+	versionKey := c.key.WithValueFlag().WithFieldId(core.DATASTORE_DOC_VERSION_FIELD_ID)
+	objectMarker, err := c.store.Get(ctx, c.key.ToPrimaryDataStoreKey().ToDS())
+	hasObjectMarker := !errors.Is(err, ds.ErrNotFound)
+	if err != nil && hasObjectMarker {
+		return err
+	}
+
+	if bytes.Equal(objectMarker, []byte{base.DeletedObjectMarker}) {
+		versionKey = versionKey.WithDeletedFlag()
+	}
+
+	err = c.store.Put(ctx, versionKey.ToDS(), []byte(c.schemaVersionKey.SchemaVersionId))
 	if err != nil {
 		return err
 	}
-	if !exists {
-		// write object marker
+
+	if !hasObjectMarker {
+		// ensure object marker exists
 		return c.store.Put(ctx, c.key.ToPrimaryDataStoreKey().ToDS(), []byte{base.ObjectMarker})
 	}