feat: Add lens migration engine to defra

.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})
 	}