From fb6a336f9ec2155e9cea6dae7fdd575439c30f3d Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Mon, 1 Feb 2021 11:44:37 +0100 Subject: [PATCH] Replay network subscriptions on startup --- cmd/network.db | Bin 0 -> 16384 bytes makefile | 2 +- network/dag/{bboltdag.go => bbolt_dag.go} | 132 +++------- ...dag_contract_test.go => bbolt_dag_test.go} | 228 ++++++++---------- network/dag/bbolt_payloadstore.go | 55 +++++ network/dag/bbolt_payloadstore_test.go | 52 ++++ network/dag/bboltdag_test.go | 120 --------- network/dag/bfswalker.go | 101 ++++++++ network/dag/bfswalker_test.go | 105 ++++++++ network/dag/dag_test.go | 122 ---------- network/dag/{dag.go => interface.go} | 108 ++------- network/dag/mock.go | 164 ++++++++++--- network/dag/publisher.go | 89 +++++++ network/dag/publisher_test.go | 165 +++++++++++++ network/dag/test.go | 35 +++ network/network.go | 15 +- network/network_integration_test.go | 9 +- network/network_test.go | 10 +- network/proto/handlers.go | 4 +- 19 files changed, 910 insertions(+), 606 deletions(-) create mode 100644 cmd/network.db rename network/dag/{bboltdag.go => bbolt_dag.go} (75%) rename network/dag/{dag_contract_test.go => bbolt_dag_test.go} (51%) create mode 100644 network/dag/bbolt_payloadstore.go create mode 100644 network/dag/bbolt_payloadstore_test.go delete mode 100644 network/dag/bboltdag_test.go create mode 100644 network/dag/bfswalker.go create mode 100644 network/dag/bfswalker_test.go delete mode 100644 network/dag/dag_test.go rename network/dag/{dag.go => interface.go} (55%) create mode 100644 network/dag/publisher.go create mode 100644 network/dag/publisher_test.go diff --git a/cmd/network.db b/cmd/network.db new file mode 100644 index 0000000000000000000000000000000000000000..e449c28987ce16bf4ca8ea054d2f468599fb2e03 GIT binary patch literal 16384 zcmeI&yA8rH5C-7zkZ6#AmI2s+6%YkY7D=BGSdksV21q<|{{VDR^rbobY)ihcvz?|H zzjejou^pD*_I^yPDxdHh`*bot`RD6?IdttNeUShG0t5&UAV7cs0RjXF5SWg@qPCLr z{J-b_tlwSx_ISRXrc;@n5FkK+009C72oNAZfB*pk5y*RfJ@+Rq5+Fc;009C72oNAZ lfB*pkGZM)E{kaF2QE5scK!5-N0t5&UAV7cs0RjX{;0=f~3rhe1 literal 0 HcmV?d00001 diff --git a/makefile b/makefile index 2bccb1bb19..4c7b4ab820 100644 --- a/makefile +++ b/makefile @@ -12,7 +12,7 @@ gen-mocks: mockgen -destination=network/proto/mock.go -package=proto -source=network/proto/interface.go Protocol mockgen -destination=network/p2p/mock.go -package=p2p -source=network/p2p/interface.go P2PNetwork mockgen -destination=network/mock.go -package=network -source=network/interface.go - mockgen -destination=network/dag/mock.go -package=dag -source=network/dag/dag.go DAG PayloadStore + mockgen -destination=network/dag/mock.go -package=dag -source=network/dag/interface.go DAG PayloadStore gen-api: oapi-codegen -generate types,server,client -package v1 docs/_static/crypto/v1.yaml > crypto/api/v1/generated.go diff --git a/network/dag/bboltdag.go b/network/dag/bbolt_dag.go similarity index 75% rename from network/dag/bboltdag.go rename to network/dag/bbolt_dag.go index ecd4500f7c..9aa7b5d61f 100644 --- a/network/dag/bboltdag.go +++ b/network/dag/bbolt_dag.go @@ -27,9 +27,6 @@ import ( "go.etcd.io/bbolt" ) -// payloadsBucket is the name of the Bolt bucket that holds the payloads of the documents. -const payloadsBucket = "payloads" - // documentsBucket is the name of the Bolt bucket that holds the actual documents as JSON. const documentsBucket = "documents" @@ -52,12 +49,9 @@ const rootsDocumentKey = "roots" // headsBucket contains the name of the bucket the holds the heads. const headsBucket = "heads" -// boltDBFileMode holds the Unix file mode the created BBolt database files will have. -const boltDBFileMode = 0600 - type bboltDAG struct { - db *bbolt.DB - subscribers map[string]Receiver + db *bbolt.DB + observers []Observer } type headsStatistic struct { @@ -97,19 +91,13 @@ func (d dataSizeStatistic) String() string { return fmt.Sprintf("%d", d.sizeInBytes) } -// NewBBoltDAG creates a etcd/bbolt backed DAG using the given database file path. If the file doesn't exist, it's created. -// The parent directory of the path must exist, otherwise an error could be returned. If the file can't be created or -// read, an error is returned as well. -func NewBBoltDAG(path string) (DAG, PayloadStore, error) { - db, err := bbolt.Open(path, boltDBFileMode, bbolt.DefaultOptions) - if err != nil { - return nil, nil, fmt.Errorf("unable to create bbolt DAG: %w", err) - } - instance := &bboltDAG{ - db: db, - subscribers: map[string]Receiver{}, - } - return instance, instance, nil +// NewBBoltDAG creates a etcd/bbolt backed DAG using the given database. +func NewBBoltDAG(db *bbolt.DB) DAG { + return &bboltDAG{db: db} +} + +func (dag *bboltDAG) RegisterObserver(observer Observer) { + dag.observers = append(dag.observers, observer) } func (dag *bboltDAG) Diagnostics() []core.DiagnosticResult { @@ -128,47 +116,6 @@ func (dag *bboltDAG) Diagnostics() []core.DiagnosticResult { return result } -func (dag *bboltDAG) Subscribe(documentType string, receiver Receiver) { - oldSubscriber := dag.subscribers[documentType] - dag.subscribers[documentType] = func(document Document, payload []byte) error { - // Chain subscribers in case there's more than 1 - if oldSubscriber != nil { - if err := oldSubscriber(document, payload); err != nil { - return err - } - } - return receiver(document, payload) - } -} - -func (dag bboltDAG) ReadPayload(payloadHash hash.SHA256Hash) ([]byte, error) { - var result []byte - err := dag.db.View(func(tx *bbolt.Tx) error { - if payloads := tx.Bucket([]byte(payloadsBucket)); payloads != nil { - result = payloads.Get(payloadHash.Slice()) - } - return nil - }) - return result, err -} - -func (dag bboltDAG) WritePayload(payloadHash hash.SHA256Hash, data []byte) error { - err := dag.db.Update(func(tx *bbolt.Tx) error { - payloads, err := tx.CreateBucketIfNotExists([]byte(payloadsBucket)) - if err != nil { - return err - } - if err := payloads.Put(payloadHash.Slice(), data); err != nil { - return err - } - return nil - }) - if err == nil { - dag.payloadReceived(payloadHash, data) - } - return err -} - func (dag bboltDAG) Get(ref hash.SHA256Hash) (Document, error) { var result Document var err error @@ -242,11 +189,7 @@ func (dag bboltDAG) All() ([]Document, error) { } func (dag bboltDAG) IsPresent(ref hash.SHA256Hash) (bool, error) { - return dag.isPresent(documentsBucket, ref.Slice()) -} - -func (dag bboltDAG) IsPayloadPresent(payloadHash hash.SHA256Hash) (bool, error) { - return dag.isPresent(payloadsBucket, payloadHash.Slice()) + return isPresent(dag.db, documentsBucket, ref.Slice()) } func (dag bboltDAG) MissingDocuments() []hash.SHA256Hash { @@ -276,7 +219,7 @@ func (dag *bboltDAG) Add(documents ...Document) error { return nil } -func (dag bboltDAG) Walk(walker Walker, visitor Visitor, startAt hash.SHA256Hash) error { +func (dag bboltDAG) Walk(algo WalkerAlgorithm, visitor Visitor, startAt hash.SHA256Hash) error { return dag.db.View(func(tx *bbolt.Tx) error { documents := tx.Bucket([]byte(documentsBucket)) nexts := tx.Bucket([]byte(nextsBucket)) @@ -284,11 +227,11 @@ func (dag bboltDAG) Walk(walker Walker, visitor Visitor, startAt hash.SHA256Hash // DAG is empty return nil } - return walker.walk(visitor, startAt, func(hash hash.SHA256Hash) (Document, error) { + return algo.walk(visitor, startAt, func(hash hash.SHA256Hash) (Document, error) { return getDocument(hash, documents) }, func(hash hash.SHA256Hash) ([]hash.SHA256Hash, error) { return parseHashList(nexts.Get(hash.Slice())), nil - }, documents.Stats().KeyN) // TODO Optimization: should we cache this number of keys? + }) }) } @@ -304,10 +247,10 @@ func (dag bboltDAG) Root() (hash hash.SHA256Hash, err error) { return } -func (dag bboltDAG) isPresent(bucketName string, key []byte) (bool, error) { +func isPresent(db *bbolt.DB, bucketName string, key []byte) (bool, error) { var result bool var err error - err = dag.db.View(func(tx *bbolt.Tx) error { + err = db.View(func(tx *bbolt.Tx) error { if payloads := tx.Bucket([]byte(bucketName)); payloads != nil { data := payloads.Get(key) result = len(data) > 0 @@ -320,8 +263,8 @@ func (dag bboltDAG) isPresent(bucketName string, key []byte) (bool, error) { func (dag *bboltDAG) add(document Document) error { ref := document.Ref() refSlice := ref.Slice() - return dag.db.Update(func(tx *bbolt.Tx) error { - documents, nexts, missingDocuments, payloadIndex, _, heads, err := getBuckets(tx) + err := dag.db.Update(func(tx *bbolt.Tx) error { + documents, nexts, missingDocuments, payloadIndex, heads, err := getBuckets(tx) if err != nil { return err } @@ -370,9 +313,13 @@ func (dag *bboltDAG) add(document Document) error { // Remove marker if this document was previously missing return missingDocuments.Delete(refSlice) }) + if err == nil { + notifyObservers(dag.observers, document) + } + return err } -func getBuckets(tx *bbolt.Tx) (documents, nexts, missingDocuments, payloadIndex, payloads, heads *bbolt.Bucket, err error) { +func getBuckets(tx *bbolt.Tx) (documents, nexts, missingDocuments, payloadIndex, heads *bbolt.Bucket, err error) { if documents, err = tx.CreateBucketIfNotExists([]byte(documentsBucket)); err != nil { return } @@ -385,9 +332,6 @@ func getBuckets(tx *bbolt.Tx) (documents, nexts, missingDocuments, payloadIndex, if payloadIndex, err = tx.CreateBucketIfNotExists([]byte(payloadIndexBucket)); err != nil { return } - if payloads, err = tx.CreateBucketIfNotExists([]byte(payloadsBucket)); err != nil { - return - } if heads, err = tx.CreateBucketIfNotExists([]byte(headsBucket)); err != nil { return } @@ -416,32 +360,6 @@ func (dag *bboltDAG) registerNextRef(nextsBucket *bbolt.Bucket, prev hash.SHA256 return nextsBucket.Put(prevSlice, appendHashList(value, next)) } -func (dag *bboltDAG) payloadReceived(payloadHash hash.SHA256Hash, payload []byte) { - // TODO: This is a stupid implementation that doesn't retry failed subscribers or publish documents in-order - // (since documents and payload may arrive out-of-order). Should be changed to something more intelligent. - err := dag.db.View(func(tx *bbolt.Tx) error { - documents := tx.Bucket([]byte(documentsBucket)) - payloadsIndex := tx.Bucket([]byte(payloadIndexBucket)) - if documents == nil || payloadsIndex == nil { - return nil - } - for _, documentRef := range parseHashList(payloadsIndex.Get(payloadHash.Slice())) { - if document, err := getDocument(documentRef, documents); err != nil { - return err - } else if receiver := dag.subscribers[document.PayloadType()]; receiver == nil { - continue - } else if err := receiver(document, payload); err != nil { - // TODO: Should this be done in a goroutine to make sure applications don't block network processes? - log.Logger().Errorf("Document subscriber returned an error (document=%s,type=%s): %v", document.Ref(), document.PayloadType(), err) - } - } - return nil - }) - if err != nil { - log.Logger().Errorf("Unable to publish document (payload=%s): %v", payloadHash, err) - } -} - func getDocument(hash hash.SHA256Hash, documents *bbolt.Bucket) (Document, error) { documentBytes := documents.Get(hash.Slice()) if documentBytes == nil { @@ -478,3 +396,9 @@ func appendHashList(list []byte, h hash.SHA256Hash) []byte { newList = append(newList, h.Slice()...) return newList } + +func notifyObservers(observers []Observer, subject interface{}) { + for _, observer := range observers { + observer(subject) + } +} diff --git a/network/dag/dag_contract_test.go b/network/dag/bbolt_dag_test.go similarity index 51% rename from network/dag/dag_contract_test.go rename to network/dag/bbolt_dag_test.go index 33d45b190c..af4c07bd81 100644 --- a/network/dag/dag_contract_test.go +++ b/network/dag/bbolt_dag_test.go @@ -19,13 +19,29 @@ package dag import ( - "errors" "github.com/nuts-foundation/nuts-node/crypto/hash" + "github.com/nuts-foundation/nuts-node/test/io" "github.com/stretchr/testify/assert" + "go.etcd.io/bbolt" + "path" + "sort" "strings" "testing" ) +func createBBoltDB(testDirectory string) *bbolt.DB { + db, err := bbolt.Open(path.Join(testDirectory, "dag.db"), 0600, bbolt.DefaultOptions) + if err != nil { + panic(err) + } + return db +} + +func createDAG(t *testing.T) DAG { + testDirectory := io.TestDirectory(t) + return NewBBoltDAG(createBBoltDB(testDirectory)) +} + // trackingVisitor just keeps track of which nodes were visited in what order. type trackingVisitor struct { documents []Document @@ -48,9 +64,9 @@ func (n trackingVisitor) JoinRefsAsString() string { return strings.Join(contents, ", ") } -func DAGTest_All(creator func(t *testing.T) DAG, t *testing.T) { +func TestBBoltDAG_All(t *testing.T) { t.Run("ok", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) doc := CreateTestDocumentWithJWK(1) err := graph.Add(doc) @@ -68,9 +84,9 @@ func DAGTest_All(creator func(t *testing.T) DAG, t *testing.T) { }) } -func DAGTest_Get(creator func(t *testing.T) DAG, t *testing.T) { +func TestBBoltDAG_Get(t *testing.T) { t.Run("found", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) document := CreateTestDocumentWithJWK(1) _ = graph.Add(document) actual, err := graph.Get(document.Ref()) @@ -80,16 +96,16 @@ func DAGTest_Get(creator func(t *testing.T) DAG, t *testing.T) { assert.Equal(t, document, actual) }) t.Run("not found", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) actual, err := graph.Get(hash.SHA256Sum([]byte{1, 2, 3})) assert.NoError(t, err) assert.Nil(t, actual) }) } -func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { +func TestBBoltDAG_Add(t *testing.T) { t.Run("ok", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) doc := CreateTestDocumentWithJWK(0) err := graph.Add(doc) @@ -97,7 +113,7 @@ func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { assert.NoError(t, err) visitor := trackingVisitor{} root, _ := graph.Root() - err = graph.Walk(&BFSWalker{}, visitor.Accept, root) + err = graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) if !assert.NoError(t, err) { return } @@ -107,7 +123,7 @@ func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { assert.True(t, present) }) t.Run("duplicate", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) doc := CreateTestDocumentWithJWK(0) _ = graph.Add(doc) @@ -117,7 +133,7 @@ func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { assert.Len(t, actual, 1) }) t.Run("second root", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) root1 := CreateTestDocumentWithJWK(1) root2 := CreateTestDocumentWithJWK(2) @@ -128,8 +144,8 @@ func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { assert.Len(t, actual, 1) }) t.Run("ok - out of order", func(t *testing.T) { - _, documents := graphF(creator, t) - graph := creator(t) + graph := createDAG(t) + documents := graphF() for i := len(documents) - 1; i >= 0; i-- { err := graph.Add(documents[i]) @@ -140,7 +156,7 @@ func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { visitor := trackingVisitor{} root, _ := graph.Root() - err := graph.Walk(&BFSWalker{}, visitor.Accept, root) + err := graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) if !assert.NoError(t, err) { return } @@ -154,19 +170,19 @@ func DAGTest_Add(creator func(t *testing.T) DAG, t *testing.T) { C := CreateTestDocumentWithJWK(2, B.Ref()) B.prevs = append(B.prevs, C.Ref()) - graph := creator(t) + graph := createDAG(t) err := graph.Add(A, B, C) assert.EqualError(t, err, "") }) } -func DAGTest_Walk(creator func(t *testing.T) DAG, t *testing.T) { +func TestBBoltDAG_Walk(t *testing.T) { t.Run("ok - empty graph", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) visitor := trackingVisitor{} root, _ := graph.Root() - err := graph.Walk(&BFSWalker{}, visitor.Accept, root) + err := graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) if !assert.NoError(t, err) { return } @@ -175,21 +191,21 @@ func DAGTest_Walk(creator func(t *testing.T) DAG, t *testing.T) { }) } -func DAGTest_MissingDocuments(creator func(t *testing.T) DAG, t *testing.T) { +func TestBBoltDAG_MissingDocuments(t *testing.T) { A := CreateTestDocumentWithJWK(0) B := CreateTestDocumentWithJWK(1, A.Ref()) C := CreateTestDocumentWithJWK(2, B.Ref()) t.Run("no missing documents (empty graph)", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) assert.Empty(t, graph.MissingDocuments()) }) t.Run("no missing documents (non-empty graph)", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) graph.Add(A, B, C) assert.Empty(t, graph.MissingDocuments()) }) t.Run("missing documents (non-empty graph)", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) graph.Add(A, C) assert.Len(t, graph.MissingDocuments(), 1) // Now add missing document B and assert there are no more missing documents @@ -197,75 +213,21 @@ func DAGTest_MissingDocuments(creator func(t *testing.T) DAG, t *testing.T) { assert.Empty(t, graph.MissingDocuments()) }) } - -func DAGTest_Subscribe(creator func(t *testing.T) DAG, t *testing.T) { - t.Run("no subscribers", func(t *testing.T) { - graph := creator(t) - document := CreateTestDocumentWithJWK(1) - _ = graph.Add(document) - err := graph.(PayloadStore).WritePayload(document.Payload(), []byte("foobar")) - assert.NoError(t, err) - }) - t.Run("single subscriber", func(t *testing.T) { - graph := creator(t) - document := CreateTestDocumentWithJWK(1) - received := false - graph.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { - assert.Equal(t, document, actualDocument) - assert.Equal(t, []byte("foobar"), actualPayload) - received = true - return nil - }) - _ = graph.Add(document) - err := graph.(PayloadStore).WritePayload(document.Payload(), []byte("foobar")) - assert.NoError(t, err) - assert.True(t, received) - }) - t.Run("multiple subscribers", func(t *testing.T) { - graph := creator(t) - document := CreateTestDocumentWithJWK(1) - calls := 0 - receiver := func(actualDocument Document, actualPayload []byte) error { - calls++ - return nil - } - graph.Subscribe(document.PayloadType(), receiver) - graph.Subscribe(document.PayloadType(), receiver) - _ = graph.Add(document) - err := graph.(PayloadStore).WritePayload(document.Payload(), []byte("foobar")) - assert.NoError(t, err) - assert.Equal(t, 2, calls) - }) - t.Run("multiple subscribers, first fails", func(t *testing.T) { - graph := creator(t) - document := CreateTestDocumentWithJWK(1) - calls := 0 - receiver := func(actualDocument Document, actualPayload []byte) error { - calls++ - return errors.New("failed") - } - graph.Subscribe(document.PayloadType(), receiver) - graph.Subscribe(document.PayloadType(), receiver) - _ = graph.Add(document) - err := graph.(PayloadStore).WritePayload(document.Payload(), []byte("foobar")) - assert.NoError(t, err) - assert.Equal(t, 1, calls) - }) - t.Run("subscriber error", func(t *testing.T) { - graph := creator(t) - document := CreateTestDocumentWithJWK(1) - graph.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { - return errors.New("failed") - }) - _ = graph.Add(document) - err := graph.(PayloadStore).WritePayload(document.Payload(), []byte("foobar")) - assert.NoError(t, err) +func TestBBoltDAG_Observe(t *testing.T) { + graph := createDAG(t) + var actual interface{} + graph.RegisterObserver(func(subject interface{}) { + actual = subject }) + expected := CreateTestDocumentWithJWK(1) + err := graph.Add(expected) + assert.NoError(t, err) + assert.Equal(t, expected, actual) } -func DAGTest_GetByPayloadHash(creator func(t *testing.T) DAG, t *testing.T) { +func TestBBoltDAG_GetByPayloadHash(t *testing.T) { t.Run("found", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) document := CreateTestDocumentWithJWK(1) _ = graph.Add(document) actual, err := graph.GetByPayloadHash(document.Payload()) @@ -274,59 +236,57 @@ func DAGTest_GetByPayloadHash(creator func(t *testing.T) DAG, t *testing.T) { assert.Equal(t, document, actual[0]) }) t.Run("not found", func(t *testing.T) { - graph := creator(t) + graph := createDAG(t) actual, err := graph.GetByPayloadHash(hash.SHA256Sum([]byte{1, 2, 3})) assert.NoError(t, err) assert.Empty(t, actual) }) } -func PayloadStoreTest(creator func(t *testing.T) PayloadStore, t *testing.T) { - t.Run("roundtrip", func(t *testing.T) { - payload := []byte("Hello, World!") - hash := hash.SHA256Sum(payload) - payloadStore := creator(t) - // Before, payload should not be present - present, err := payloadStore.IsPayloadPresent(hash) - if !assert.NoError(t, err) || !assert.False(t, present) { - return - } - // Add payload - err = payloadStore.WritePayload(hash, payload) - if !assert.NoError(t, err) { - return - } - // Now it should be present - present, err = payloadStore.IsPayloadPresent(hash) - if !assert.NoError(t, err) || !assert.True(t, present, "payload should be present") { - return - } - // Read payload - data, err := payloadStore.ReadPayload(hash) - if !assert.NoError(t, err) { - return - } - assert.Equal(t, payload, data) - }) +func TestBBoltDAG_Diagnostics(t *testing.T) { + dag := createDAG(t).(*bboltDAG) + doc1 := CreateTestDocumentWithJWK(2) + dag.Add(doc1) + diagnostics := dag.Diagnostics() + assert.Len(t, diagnostics, 3) + // Assert actual diagnostics + lines := make([]string, 0) + for _, diagnostic := range diagnostics { + lines = append(lines, diagnostic.Name()+": "+diagnostic.String()) + } + sort.Strings(lines) + actual := strings.Join(lines, "\n") + assert.Equal(t, `[DAG] Heads: [`+doc1.Ref().String()+`] +[DAG] Number of documents: 2 +[DAG] Stored document size (bytes): 0`, actual) } -// graphF creates the following graph: -//..................A -//................/ \ -//...............B C -//...............\ / \ -//.................D E -//.......................\ -//........................F -func graphF(creator func(t *testing.T) DAG, t *testing.T) (DAG, []Document) { - graph := creator(t) - A := CreateTestDocumentWithJWK(0) - B := CreateTestDocumentWithJWK(1, A.Ref()) - C := CreateTestDocumentWithJWK(2, A.Ref()) - D := CreateTestDocumentWithJWK(3, B.Ref(), C.Ref()) - E := CreateTestDocumentWithJWK(4, C.Ref()) - F := CreateTestDocumentWithJWK(5, E.Ref()) - docs := []Document{A, B, C, D, E, F} - graph.Add(docs...) - return graph, docs +func Test_parseHashList(t *testing.T) { + t.Run("empty", func(t *testing.T) { + assert.Empty(t, parseHashList([]byte{})) + }) + t.Run("1 entry", func(t *testing.T) { + h1 := hash.SHA256Sum([]byte("Hello, World!")) + actual := parseHashList(h1[:]) + assert.Len(t, actual, 1) + assert.Equal(t, hash.FromSlice(h1[:]), actual[0]) + }) + t.Run("2 entries", func(t *testing.T) { + h1 := hash.SHA256Sum([]byte("Hello, World!")) + h2 := hash.SHA256Sum([]byte("Hello, All!")) + actual := parseHashList(append(h1[:], h2[:]...)) + assert.Len(t, actual, 2) + assert.Equal(t, hash.FromSlice(h1[:]), actual[0]) + assert.Equal(t, hash.FromSlice(h2[:]), actual[1]) + }) + t.Run("2 entries, dangling bytes", func(t *testing.T) { + h1 := hash.SHA256Sum([]byte("Hello, World!")) + h2 := hash.SHA256Sum([]byte("Hello, All!")) + input := append(h1[:], h2[:]...) + input = append(input, 1, 2, 3) // Add some dangling bytes + actual := parseHashList(input) + assert.Len(t, actual, 2) + assert.Equal(t, hash.FromSlice(h1[:]), actual[0]) + assert.Equal(t, hash.FromSlice(h2[:]), actual[1]) + }) } diff --git a/network/dag/bbolt_payloadstore.go b/network/dag/bbolt_payloadstore.go new file mode 100644 index 0000000000..96c22df4ac --- /dev/null +++ b/network/dag/bbolt_payloadstore.go @@ -0,0 +1,55 @@ +package dag + +import ( + "github.com/nuts-foundation/nuts-node/crypto/hash" + "go.etcd.io/bbolt" +) + +// payloadsBucket is the name of the Bolt bucket that holds the payloads of the documents. +const payloadsBucket = "payloads" + +// NewBBoltPayloadStore creates a etcd/bbolt backed payload store using the given database. +func NewBBoltPayloadStore(db *bbolt.DB) PayloadStore { + return &bboltPayloadStore{db: db, observers: []Observer{}} +} + +type bboltPayloadStore struct { + db *bbolt.DB + observers []Observer +} + +func (store *bboltPayloadStore) RegisterObserver(observer Observer) { + store.observers = append(store.observers, observer) +} + +func (store bboltPayloadStore) IsPresent(payloadHash hash.SHA256Hash) (bool, error) { + return isPresent(store.db, payloadsBucket, payloadHash.Slice()) +} + +func (store bboltPayloadStore) ReadPayload(payloadHash hash.SHA256Hash) ([]byte, error) { + var result []byte + err := store.db.View(func(tx *bbolt.Tx) error { + if payloads := tx.Bucket([]byte(payloadsBucket)); payloads != nil { + result = payloads.Get(payloadHash.Slice()) + } + return nil + }) + return result, err +} + +func (store bboltPayloadStore) WritePayload(payloadHash hash.SHA256Hash, data []byte) error { + err := store.db.Update(func(tx *bbolt.Tx) error { + payloads, err := tx.CreateBucketIfNotExists([]byte(payloadsBucket)) + if err != nil { + return err + } + if err := payloads.Put(payloadHash.Slice(), data); err != nil { + return err + } + return nil + }) + if err == nil { + notifyObservers(store.observers, payloadHash) + } + return err +} diff --git a/network/dag/bbolt_payloadstore_test.go b/network/dag/bbolt_payloadstore_test.go new file mode 100644 index 0000000000..807c7b5ecc --- /dev/null +++ b/network/dag/bbolt_payloadstore_test.go @@ -0,0 +1,52 @@ +package dag + +import ( + "github.com/nuts-foundation/nuts-node/crypto/hash" + "github.com/nuts-foundation/nuts-node/test/io" + "github.com/stretchr/testify/assert" + "testing" +) + +func TestBBoltPayloadStore_ReadWrite(t *testing.T) { + testDirectory := io.TestDirectory(t) + payloadStore := NewBBoltPayloadStore(createBBoltDB(testDirectory)) + + payload := []byte("Hello, World!") + hash := hash.SHA256Sum(payload) + // Before, payload should not be present + present, err := payloadStore.IsPresent(hash) + if !assert.NoError(t, err) || !assert.False(t, present) { + return + } + // Add payload + err = payloadStore.WritePayload(hash, payload) + if !assert.NoError(t, err) { + return + } + // Now it should be present + present, err = payloadStore.IsPresent(hash) + if !assert.NoError(t, err) || !assert.True(t, present, "payload should be present") { + return + } + // Read payload + data, err := payloadStore.ReadPayload(hash) + if !assert.NoError(t, err) { + return + } + assert.Equal(t, payload, data) +} + +func TestBBoltPayloadStore_Observe(t *testing.T) { + testDirectory := io.TestDirectory(t) + payloadStore := NewBBoltPayloadStore(createBBoltDB(testDirectory)) + + var actual interface{} + payloadStore.RegisterObserver(func(subject interface{}) { + actual = subject + }) + payload := []byte(t.Name()) + expected := hash.SHA256Sum(payload) + err := payloadStore.WritePayload(expected, payload) + assert.NoError(t, err) + assert.Equal(t, expected, actual) +} diff --git a/network/dag/bboltdag_test.go b/network/dag/bboltdag_test.go deleted file mode 100644 index 93e7957061..0000000000 --- a/network/dag/bboltdag_test.go +++ /dev/null @@ -1,120 +0,0 @@ -/* - * Copyright (C) 2021. Nuts community - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . - * - */ - -package dag - -import ( - "github.com/nuts-foundation/nuts-node/crypto/hash" - "github.com/nuts-foundation/nuts-node/test/io" - "github.com/stretchr/testify/assert" - "path" - "sort" - "strings" - "testing" -) - -var bboltDAGCreator = func(t *testing.T) DAG { - if dag, _, err := NewBBoltDAG(path.Join(io.TestDirectory(t), "dag.db")); err != nil { - t.Fatal(err) - return nil - } else { - return dag - } -} - -func TestBBoltDAG_Add(t *testing.T) { - DAGTest_Add(bboltDAGCreator, t) -} - -func TestBBoltDAG_All(t *testing.T) { - DAGTest_All(bboltDAGCreator, t) -} - -func TestBBoltDAG_MissingDocuments(t *testing.T) { - DAGTest_MissingDocuments(bboltDAGCreator, t) -} - -func TestBBoltDAG_Walk(t *testing.T) { - DAGTest_Walk(bboltDAGCreator, t) -} - -func TestBBoltDAG_Get(t *testing.T) { - DAGTest_Get(bboltDAGCreator, t) -} - -func TestBBoltDAG_GetByPayloadHash(t *testing.T) { - DAGTest_GetByPayloadHash(bboltDAGCreator, t) -} - -func TestBBoltDAG_PayloadStore(t *testing.T) { - PayloadStoreTest(func(t *testing.T) PayloadStore { - return bboltDAGCreator(t).(PayloadStore) - }, t) -} - -func TestBBoltDAG_Subscribe(t *testing.T) { - DAGTest_Subscribe(bboltDAGCreator, t) -} - -func TestBBoltDAG_Diagnostics(t *testing.T) { - dag := bboltDAGCreator(t).(*bboltDAG) - doc1 := CreateTestDocumentWithJWK(2) - dag.Add(doc1) - diagnostics := dag.Diagnostics() - assert.Len(t, diagnostics, 3) - // Assert actual diagnostics - lines := make([]string, 0) - for _, diagnostic := range diagnostics { - lines = append(lines, diagnostic.Name()+": "+diagnostic.String()) - } - sort.Strings(lines) - actual := strings.Join(lines, "\n") - assert.Equal(t, `[DAG] Heads: [`+doc1.Ref().String()+`] -[DAG] Number of documents: 2 -[DAG] Stored document size (bytes): 0`, actual) -} - -func Test_parseHashList(t *testing.T) { - t.Run("empty", func(t *testing.T) { - assert.Empty(t, parseHashList([]byte{})) - }) - t.Run("1 entry", func(t *testing.T) { - h1 := hash.SHA256Sum([]byte("Hello, World!")) - actual := parseHashList(h1[:]) - assert.Len(t, actual, 1) - assert.Equal(t, hash.FromSlice(h1[:]), actual[0]) - }) - t.Run("2 entries", func(t *testing.T) { - h1 := hash.SHA256Sum([]byte("Hello, World!")) - h2 := hash.SHA256Sum([]byte("Hello, All!")) - actual := parseHashList(append(h1[:], h2[:]...)) - assert.Len(t, actual, 2) - assert.Equal(t, hash.FromSlice(h1[:]), actual[0]) - assert.Equal(t, hash.FromSlice(h2[:]), actual[1]) - }) - t.Run("2 entries, dangling bytes", func(t *testing.T) { - h1 := hash.SHA256Sum([]byte("Hello, World!")) - h2 := hash.SHA256Sum([]byte("Hello, All!")) - input := append(h1[:], h2[:]...) - input = append(input, 1, 2, 3) // Add some dangling bytes - actual := parseHashList(input) - assert.Len(t, actual, 2) - assert.Equal(t, hash.FromSlice(h1[:]), actual[0]) - assert.Equal(t, hash.FromSlice(h2[:]), actual[1]) - }) -} diff --git a/network/dag/bfswalker.go b/network/dag/bfswalker.go new file mode 100644 index 0000000000..fcf07d003a --- /dev/null +++ b/network/dag/bfswalker.go @@ -0,0 +1,101 @@ +package dag + +import ( + "container/list" + "github.com/nuts-foundation/nuts-node/crypto/hash" + "sort" +) + +// bfsWalkerAlgorithm walks the DAG using the Breadth-First-Search (BFS) as described by Anany Levitin in "The Design & Analysis of Algorithms". +// It visits the whole tree level for level (breadth first vs depth first). It works by taking a node from queue and +// then adds the node's children (downward edges) to the queue. It starts by adding the root node to the queue and +// loops over the queue until empty, meaning all nodes reachable from the root node have been visited. Since our +// DAG contains merges (two parents referring to the same child node) we also keep a map to avoid visiting a +// merger node twice. +// +// This also means we have to make sure we don't visit the merger node before all of its previous nodes have been +// visited, which BFS doesn't account for. If that happens we skip the node without marking it as visited, +// so it will be visited again when the unvisited previous node is visited, which re-adds the merger node to the queue. +// +// In addition, when the visitor stops the walking (by returning false), it breaks off the walking for that specific branch of the DAG. +// +// It is also stateful: it remembers which documents were visited and where to resume walking next time it is invoked. +// This is useful for subscriptions: the first time the DAG is walked all documents are processed, keeping track of +// which branches couldn't be processed (payload might be missing, so it has to be processed layer). Next time the +// walker is invoked it starts walking at the documents in the `resumeAt` list and ultimately honoring the `startAt` +// hash. +type bfsWalkerAlgorithm struct { + resumeAt *list.List + visitedDocuments map[hash.SHA256Hash]bool +} + +// NewBFSWalkerAlgorithm creates a new bfsWalkerAlgorithm. +func NewBFSWalkerAlgorithm() WalkerAlgorithm { + return &bfsWalkerAlgorithm{ + resumeAt: list.New(), + visitedDocuments: map[hash.SHA256Hash]bool{}, + } +} + +func (w bfsWalkerAlgorithm) walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(hash.SHA256Hash) (Document, error), nextsFn func(hash.SHA256Hash) ([]hash.SHA256Hash, error)) error { + queue := list.New() + queue.PushFrontList(w.resumeAt) + if !startAt.Empty() { + queue.PushBack(startAt) + } + w.resumeAt.Init() +ProcessQueueLoop: + for queue.Len() > 0 { + // Pop first element of queue + front := queue.Front() + queue.Remove(front) + currentRef := front.Value.(hash.SHA256Hash) + + // Make sure we haven't already visited this node + if _, visited := w.visitedDocuments[currentRef]; visited { + continue + } + + // Make sure all prevs have been visited. Otherwise just continue, it will be re-added to the queue when the + // unvisited prev node is visited and re-adds this node to the processing queue. + currentDocument, err := getFn(currentRef) + if err != nil { + return err + } + + for _, prev := range currentDocument.Previous() { + if _, visited := w.visitedDocuments[prev]; !visited { + continue ProcessQueueLoop + } + } + + // Visit the node + if !visitor(currentDocument) { + // Visitor returned false, so stop processing this branch. Resume at later point. + w.resumeAt.PushBack(currentRef) + continue + } + + // Add child nodes to processing queue + // Processing order of nodes on the same level doesn't really matter for correctness of the DAG travel + // but it makes testing easier. + if nexts, err := nextsFn(currentRef); err != nil { + return err + } else if nexts != nil { + sortedEdges := make([]hash.SHA256Hash, 0, len(nexts)) + for _, nextNode := range nexts { + sortedEdges = append(sortedEdges, nextNode) + } + sort.Slice(sortedEdges, func(i, j int) bool { + return sortedEdges[i].Compare(sortedEdges[j]) < 0 + }) + for _, nextNode := range sortedEdges { + queue.PushBack(nextNode) + } + } + + // Mark this node as visited + w.visitedDocuments[currentRef] = true + } + return nil +} diff --git a/network/dag/bfswalker_test.go b/network/dag/bfswalker_test.go new file mode 100644 index 0000000000..807f8a99bd --- /dev/null +++ b/network/dag/bfswalker_test.go @@ -0,0 +1,105 @@ +package dag + +import ( + "github.com/nuts-foundation/nuts-node/crypto/hash" + "github.com/stretchr/testify/assert" + "testing" +) + +func TestBFSWalkerAlgorithm(t *testing.T) { + t.Run("ok - walk graph F", func(t *testing.T) { + graph := createDAG(t) + graph.Add(graphF()...) + visitor := trackingVisitor{} + + root, _ := graph.Root() + err := graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) + if !assert.NoError(t, err) { + return + } + + assert.Regexp(t, "^0, (1, 2|2, 1), (3, 4|4, 3), 5$", visitor.JoinRefsAsString()) + }) + + t.Run("ok - walk graph G", func(t *testing.T) { + graph := createDAG(t) + graph.Add(graphG()...) + visitor := trackingVisitor{} + root, _ := graph.Root() + graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) + + assert.Regexp(t, "^0, (1, 2|2, 1), (3, 4|4, 3), 5, 6$", visitor.JoinRefsAsString()) + }) + + t.Run("ok - walk graph F, C is missing", func(t *testing.T) { + //..................A + //................/ \ + //...............B C (missing) + //...............\ / \ + //.................D E + //.......................\ + //........................F + graph := createDAG(t) + visitor := trackingVisitor{} + docs := graphF() + graph.Add(docs[0], docs[1], docs[3], docs[4], docs[5]) + + root, _ := graph.Root() + graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) + + assert.Equal(t, "0, 1", visitor.JoinRefsAsString()) + }) + + t.Run("ok - walk graph G, C resume point", func(t *testing.T) { + //..................A + //................/ \ + //...............B C (resume point) + //...............\ / \ + //.................D E + //.......................\ + //........................F + graph := createDAG(t) + docs := graphG() + graph.Add(docs...) + docC := docs[2] + + root, _ := graph.Root() + walker := NewBFSWalkerAlgorithm() + visitorBeforeResume := trackingVisitor{} + graph.Walk(walker, func(document Document) bool { + if document.Ref().Equals(docC.Ref()) { + return false + } + return visitorBeforeResume.Accept(document) + }, root) + + // Make sure it breaks at C, having processed A and B + assert.Equal(t, "0, 1", visitorBeforeResume.JoinRefsAsString()) + + // Now resume, not breaking at C + visitorAfterResume := trackingVisitor{} + graph.Walk(walker, visitorAfterResume.Accept, hash.EmptyHash()) + + // Make sure it resumes at C, then processes E, D and F. + refs := visitorAfterResume.JoinRefsAsString() + assert.Regexp(t, "^2, (4, 3|3, 4), 5, 6$", refs) + + // Make sure DAG isn't revisited after walk is invoked when the complete DAG has been walked already + visitorAfterCompleteWalk := trackingVisitor{} + graph.Walk(walker, visitorAfterResume.Accept, root) + assert.Equal(t, "", visitorAfterCompleteWalk.JoinRefsAsString()) + }) + + t.Run("ok - empty graph", func(t *testing.T) { + graph := createDAG(t) + visitor := trackingVisitor{} + + root, _ := graph.Root() + err := graph.Walk(NewBFSWalkerAlgorithm(), visitor.Accept, root) + if !assert.NoError(t, err) { + return + } + + assert.Empty(t, visitor.documents) + }) +} diff --git a/network/dag/dag_test.go b/network/dag/dag_test.go deleted file mode 100644 index ed391673d3..0000000000 --- a/network/dag/dag_test.go +++ /dev/null @@ -1,122 +0,0 @@ -/* - * Copyright (C) 2021. Nuts community - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . - * - */ - -package dag - -import ( - "github.com/stretchr/testify/assert" - "testing" -) - -func TestBFSWalker(t *testing.T) { - t.Run("ok - walk graph F", func(t *testing.T) { - visitor := trackingVisitor{} - graph, _ := graphF(bboltDAGCreator, t) - - root, _ := graph.Root() - err := graph.Walk(&BFSWalker{}, visitor.Accept, root) - if !assert.NoError(t, err) { - return - } - - assert.Regexp(t, "0, (1, 2|2, 1), (3, 4|4, 3), 5", visitor.JoinRefsAsString()) - }) - - t.Run("ok - walk graph G", func(t *testing.T) { - //..................A - //................/ \ - //...............B C - //...............\ / \ - //.................D E - //.................\.....\ - //..................\.....F - //...................\.../ - //.....................G - visitor := trackingVisitor{} - graph, docs := graphF(bboltDAGCreator, t) - G := CreateTestDocumentWithJWK(6, docs[3].Ref(), docs[5].Ref()) - graph.Add(G) - - root, _ := graph.Root() - graph.Walk(&BFSWalker{}, visitor.Accept, root) - - assert.Regexp(t, "0, (1, 2|2, 1), (3, 4|4, 3), 5, 6", visitor.JoinRefsAsString()) - }) - - t.Run("ok - walk graph F, C is missing", func(t *testing.T) { - //..................A - //................/ \ - //...............B C (missing) - //...............\ / \ - //.................D E - //.......................\ - //........................F - visitor := trackingVisitor{} - _, docs := graphF(bboltDAGCreator, t) - graph := bboltDAGCreator(t) - graph.Add(docs[0], docs[1], docs[3], docs[4], docs[5]) - - root, _ := graph.Root() - graph.Walk(&BFSWalker{}, visitor.Accept, root) - - assert.Equal(t, "0, 1", visitor.JoinRefsAsString()) - }) - - t.Run("ok - empty graph", func(t *testing.T) { - graph := bboltDAGCreator(t) - visitor := trackingVisitor{} - - root, _ := graph.Root() - err := graph.Walk(&BFSWalker{}, visitor.Accept, root) - if !assert.NoError(t, err) { - return - } - - assert.Empty(t, visitor.documents) - }) - - t.Run("ok - document added twice", func(t *testing.T) { - graph := bboltDAGCreator(t) - d := CreateTestDocumentWithJWK(0) - graph.Add(d) - graph.Add(d) - visitor := trackingVisitor{} - - root, _ := graph.Root() - graph.Walk(&BFSWalker{}, visitor.Accept, root) - - assert.Len(t, visitor.documents, 1) - }) - - t.Run("error - second root document", func(t *testing.T) { - graph := bboltDAGCreator(t) - d1 := CreateTestDocumentWithJWK(0) - d2 := CreateTestDocumentWithJWK(1) - err := graph.Add(d1) - - err = graph.Add(d2) - assert.Equal(t, errRootAlreadyExists, err) - visitor := trackingVisitor{} - - root, _ := graph.Root() - graph.Walk(&BFSWalker{}, visitor.Accept, root) - - assert.Len(t, visitor.documents, 1) - assert.Equal(t, d1.Data(), visitor.documents[0].Data()) - }) -} diff --git a/network/dag/dag.go b/network/dag/interface.go similarity index 55% rename from network/dag/dag.go rename to network/dag/interface.go index ca0cc4f71a..374fa7bc3f 100644 --- a/network/dag/dag.go +++ b/network/dag/interface.go @@ -19,24 +19,23 @@ package dag import ( - "container/list" "errors" "github.com/nuts-foundation/nuts-node/crypto/hash" - "sort" ) var errRootAlreadyExists = errors.New("root document already exists") // DAG is a directed acyclic graph consisting of nodes (documents) referring to preceding nodes. type DAG interface { - Publisher + // Observable allows observers to be notified when a document is added to the DAG. + Observable // Add adds one or more documents to the DAG. If it can't be added an error is returned. Nil entries are ignored. Add(documents ...Document) error // MissingDocuments returns the hashes of the documents we know we are missing and should still be resolved. MissingDocuments() []hash.SHA256Hash // Walk visits every node of the DAG, starting at the given hash working its way down each level until every leaf is visited. // when startAt is an empty hash, the walker starts at the root node. - Walk(walker Walker, visitor Visitor, startAt hash.SHA256Hash) error + Walk(algo WalkerAlgorithm, visitor Visitor, startAt hash.SHA256Hash) error // Root returns the root hash of the DAG. If there's no root an empty hash is returned. If an error occurs, it is returned. Root() (hash.SHA256Hash, error) // Get retrieves a specific document from the DAG. If it isn't found, nil is returned. @@ -55,108 +54,53 @@ type DAG interface { // Publisher defines the interface for types that publish Nuts Network documents. type Publisher interface { - // Subscribe lets an application subscribe to a specific type of document. When a new document is received (for the - // first time) the `receiver` function is called. + // Subscribe lets an application subscribe to a specific type of document. When a new document is received + //the `receiver` function is called. Subscribe(documentType string, receiver Receiver) + // Start starts the publisher. + Start() } // Receiver defines a function for processing documents when walking the DAG. type Receiver func(document Document, payload []byte) error -// Walker defines the interface for a type that can walk the DAG. -type Walker interface { +// WalkerAlgorithm defines the interface for a type that can walk the DAG. +type WalkerAlgorithm interface { // walk visits every node of the DAG, starting at the given start node and working down each level until every leaf is visited. // numberOfNodes is an indicative number of nodes that's expected to be visited. It's used for optimizing memory usage. // getFn is a function for reading a document from the DAG using the given ref hash. If not found nil must be returned. // nextsFn is a function for reading a document's nexts using the given ref hash. If not found nil must be returned. - walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(hash.SHA256Hash) (Document, error), nextsFn func(hash.SHA256Hash) ([]hash.SHA256Hash, error), numberOfNodes int) error -} - -// BFSWalker walks the DAG using the Breadth-First-Search (BFS) as described by Anany Levitin in "The Design & Analysis of Algorithms". -// It visits the whole tree level for level (breadth first vs depth first). It works by taking a node from queue and -// then adds the node's children (downward edges) to the queue. It starts by adding the root node to the queue and -// loops over the queue until empty, meaning all nodes reachable from the root node have been visited. Since our -// DAG contains merges (two parents referring to the same child node) we also keep a map to avoid visiting a -// merger node twice. -// -// This also means we have to make sure we don't visit the merger node before all of its previous nodes have been -// visited, which BFS doesn't account for. If that happens we skip the node without marking it as visited, -// so it will be visited again when the unvisited previous node is visited, which re-adds the merger node to the queue. -type BFSWalker struct{} - -func (w BFSWalker) walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(hash.SHA256Hash) (Document, error), nextsFn func(hash.SHA256Hash) ([]hash.SHA256Hash, error), numberOfNodes int) error { - queue := list.New() - queue.PushBack(startAt) - visitedDocuments := make(map[hash.SHA256Hash]bool, numberOfNodes) - -ProcessQueueLoop: - for queue.Len() > 0 { - // Pop first element of queue - front := queue.Front() - queue.Remove(front) - currentRef := front.Value.(hash.SHA256Hash) - - // Make sure we haven't already visited this node - if _, visited := visitedDocuments[currentRef]; visited { - continue - } - - // Make sure all prevs have been visited. Otherwise just continue, it will be re-added to the queue when the - // unvisited prev node is visited and re-adds this node to the processing queue. - currentDocument, err := getFn(currentRef) - if err != nil { - return err - } - - for _, prev := range currentDocument.Previous() { - if _, visited := visitedDocuments[prev]; !visited { - continue ProcessQueueLoop - } - } - - // Add child nodes to processing queue - // Processing order of nodes on the same level doesn't really matter for correctness of the DAG travel - // but it makes testing easier. - if nexts, err := nextsFn(currentRef); err != nil { - return err - } else if nexts != nil { - sortedEdges := make([]hash.SHA256Hash, 0, len(nexts)) - for _, nextNode := range nexts { - sortedEdges = append(sortedEdges, nextNode) - } - sort.Slice(sortedEdges, func(i, j int) bool { - return sortedEdges[i].Compare(sortedEdges[j]) < 0 - }) - for _, nextNode := range sortedEdges { - queue.PushBack(nextNode) - } - } - - // Visit the node - if !visitor(currentDocument) { - break - } - - // Mark this node as visited - visitedDocuments[currentRef] = true - } - return nil + walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(hash.SHA256Hash) (Document, error), nextsFn func(hash.SHA256Hash) ([]hash.SHA256Hash, error)) error } // Visitor defines the contract for a function that visits the DAG. If the function returns `false` it stops walking the DAG. type Visitor func(document Document) bool -// PayloadStore defines the interface for types that store document payloads. +// PayloadStore defines the interface for types that store and read document payloads. type PayloadStore interface { + // Observable allows observers to be notified when payload is written to the store. + Observable + PayloadWriter // IsPayloadPresent checks whether the contents for the given document are present. - IsPayloadPresent(payloadHash hash.SHA256Hash) (bool, error) + IsPresent(payloadHash hash.SHA256Hash) (bool, error) // ReadPayload reads the contents for the specified payload, identified by the given hash. If contents can't be found, // nil is returned. If something (else) goes wrong an error is returned. ReadPayload(payloadHash hash.SHA256Hash) ([]byte, error) +} +// PayloadWriter defines the interface for types that store document payloads. +type PayloadWriter interface { // WritePayload writes contents for the specified payload, identified by the given hash. Implementations must make // sure the hash matches the given contents. WritePayload(payloadHash hash.SHA256Hash, data []byte) error } + +// Observer defines the signature of a observer which can be called by an Observable. +type Observer func(subject interface{}) + +// Observable defines the interfaces for types that can be observed. +type Observable interface { + RegisterObserver(observer Observer) +} diff --git a/network/dag/mock.go b/network/dag/mock.go index df9cc7aa8d..b34e4e888d 100644 --- a/network/dag/mock.go +++ b/network/dag/mock.go @@ -1,5 +1,5 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: network/dag/dag.go +// Source: network/dag/interface.go // Package dag is a generated GoMock package. package dag @@ -33,16 +33,16 @@ func (m *MockDAG) EXPECT() *MockDAGMockRecorder { return m.recorder } -// Subscribe mocks base method -func (m *MockDAG) Subscribe(documentType string, receiver Receiver) { +// RegisterObserver mocks base method +func (m *MockDAG) RegisterObserver(observer Observer) { m.ctrl.T.Helper() - m.ctrl.Call(m, "Subscribe", documentType, receiver) + m.ctrl.Call(m, "RegisterObserver", observer) } -// Subscribe indicates an expected call of Subscribe -func (mr *MockDAGMockRecorder) Subscribe(documentType, receiver interface{}) *gomock.Call { +// RegisterObserver indicates an expected call of RegisterObserver +func (mr *MockDAGMockRecorder) RegisterObserver(observer interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Subscribe", reflect.TypeOf((*MockDAG)(nil).Subscribe), documentType, receiver) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterObserver", reflect.TypeOf((*MockDAG)(nil).RegisterObserver), observer) } // Add mocks base method @@ -78,17 +78,17 @@ func (mr *MockDAGMockRecorder) MissingDocuments() *gomock.Call { } // Walk mocks base method -func (m *MockDAG) Walk(walker Walker, visitor Visitor, startAt hash.SHA256Hash) error { +func (m *MockDAG) Walk(algo WalkerAlgorithm, visitor Visitor, startAt hash.SHA256Hash) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Walk", walker, visitor, startAt) + ret := m.ctrl.Call(m, "Walk", algo, visitor, startAt) ret0, _ := ret[0].(error) return ret0 } // Walk indicates an expected call of Walk -func (mr *MockDAGMockRecorder) Walk(walker, visitor, startAt interface{}) *gomock.Call { +func (mr *MockDAGMockRecorder) Walk(algo, visitor, startAt interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Walk", reflect.TypeOf((*MockDAG)(nil).Walk), walker, visitor, startAt) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Walk", reflect.TypeOf((*MockDAG)(nil).Walk), algo, visitor, startAt) } // Root mocks base method @@ -215,31 +215,43 @@ func (mr *MockPublisherMockRecorder) Subscribe(documentType, receiver interface{ return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Subscribe", reflect.TypeOf((*MockPublisher)(nil).Subscribe), documentType, receiver) } -// MockWalker is a mock of Walker interface -type MockWalker struct { +// Start mocks base method +func (m *MockPublisher) Start() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Start") +} + +// Start indicates an expected call of Start +func (mr *MockPublisherMockRecorder) Start() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockPublisher)(nil).Start)) +} + +// MockWalkerAlgorithm is a mock of WalkerAlgorithm interface +type MockWalkerAlgorithm struct { ctrl *gomock.Controller - recorder *MockWalkerMockRecorder + recorder *MockWalkerAlgorithmMockRecorder } -// MockWalkerMockRecorder is the mock recorder for MockWalker -type MockWalkerMockRecorder struct { - mock *MockWalker +// MockWalkerAlgorithmMockRecorder is the mock recorder for MockWalkerAlgorithm +type MockWalkerAlgorithmMockRecorder struct { + mock *MockWalkerAlgorithm } -// NewMockWalker creates a new mock instance -func NewMockWalker(ctrl *gomock.Controller) *MockWalker { - mock := &MockWalker{ctrl: ctrl} - mock.recorder = &MockWalkerMockRecorder{mock} +// NewMockWalkerAlgorithm creates a new mock instance +func NewMockWalkerAlgorithm(ctrl *gomock.Controller) *MockWalkerAlgorithm { + mock := &MockWalkerAlgorithm{ctrl: ctrl} + mock.recorder = &MockWalkerAlgorithmMockRecorder{mock} return mock } // EXPECT returns an object that allows the caller to indicate expected use -func (m *MockWalker) EXPECT() *MockWalkerMockRecorder { +func (m *MockWalkerAlgorithm) EXPECT() *MockWalkerAlgorithmMockRecorder { return m.recorder } // walk mocks base method -func (m *MockWalker) walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(hash.SHA256Hash) (Document, error), nextsFn func(hash.SHA256Hash) ([]hash.SHA256Hash, error), numberOfNodes int) error { +func (m *MockWalkerAlgorithm) walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(hash.SHA256Hash) (Document, error), nextsFn func(hash.SHA256Hash) ([]hash.SHA256Hash, error), numberOfNodes int) error { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "walk", visitor, startAt, getFn, nextsFn, numberOfNodes) ret0, _ := ret[0].(error) @@ -247,9 +259,9 @@ func (m *MockWalker) walk(visitor Visitor, startAt hash.SHA256Hash, getFn func(h } // walk indicates an expected call of walk -func (mr *MockWalkerMockRecorder) walk(visitor, startAt, getFn, nextsFn, numberOfNodes interface{}) *gomock.Call { +func (mr *MockWalkerAlgorithmMockRecorder) walk(visitor, startAt, getFn, nextsFn, numberOfNodes interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "walk", reflect.TypeOf((*MockWalker)(nil).walk), visitor, startAt, getFn, nextsFn, numberOfNodes) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "walk", reflect.TypeOf((*MockWalkerAlgorithm)(nil).walk), visitor, startAt, getFn, nextsFn, numberOfNodes) } // MockPayloadStore is a mock of PayloadStore interface @@ -275,19 +287,45 @@ func (m *MockPayloadStore) EXPECT() *MockPayloadStoreMockRecorder { return m.recorder } -// IsPayloadPresent mocks base method -func (m *MockPayloadStore) IsPayloadPresent(payloadHash hash.SHA256Hash) (bool, error) { +// RegisterObserver mocks base method +func (m *MockPayloadStore) RegisterObserver(observer Observer) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "RegisterObserver", observer) +} + +// RegisterObserver indicates an expected call of RegisterObserver +func (mr *MockPayloadStoreMockRecorder) RegisterObserver(observer interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterObserver", reflect.TypeOf((*MockPayloadStore)(nil).RegisterObserver), observer) +} + +// WritePayload mocks base method +func (m *MockPayloadStore) WritePayload(payloadHash hash.SHA256Hash, data []byte) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WritePayload", payloadHash, data) + ret0, _ := ret[0].(error) + return ret0 +} + +// WritePayload indicates an expected call of WritePayload +func (mr *MockPayloadStoreMockRecorder) WritePayload(payloadHash, data interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WritePayload", reflect.TypeOf((*MockPayloadStore)(nil).WritePayload), payloadHash, data) +} + +// IsPresent mocks base method +func (m *MockPayloadStore) IsPresent(payloadHash hash.SHA256Hash) (bool, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "IsPayloadPresent", payloadHash) + ret := m.ctrl.Call(m, "IsPresent", payloadHash) ret0, _ := ret[0].(bool) ret1, _ := ret[1].(error) return ret0, ret1 } -// IsPayloadPresent indicates an expected call of IsPayloadPresent -func (mr *MockPayloadStoreMockRecorder) IsPayloadPresent(payloadHash interface{}) *gomock.Call { +// IsPresent indicates an expected call of IsPresent +func (mr *MockPayloadStoreMockRecorder) IsPresent(payloadHash interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsPayloadPresent", reflect.TypeOf((*MockPayloadStore)(nil).IsPayloadPresent), payloadHash) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsPresent", reflect.TypeOf((*MockPayloadStore)(nil).IsPresent), payloadHash) } // ReadPayload mocks base method @@ -305,8 +343,31 @@ func (mr *MockPayloadStoreMockRecorder) ReadPayload(payloadHash interface{}) *go return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadPayload", reflect.TypeOf((*MockPayloadStore)(nil).ReadPayload), payloadHash) } +// MockPayloadWriter is a mock of PayloadWriter interface +type MockPayloadWriter struct { + ctrl *gomock.Controller + recorder *MockPayloadWriterMockRecorder +} + +// MockPayloadWriterMockRecorder is the mock recorder for MockPayloadWriter +type MockPayloadWriterMockRecorder struct { + mock *MockPayloadWriter +} + +// NewMockPayloadWriter creates a new mock instance +func NewMockPayloadWriter(ctrl *gomock.Controller) *MockPayloadWriter { + mock := &MockPayloadWriter{ctrl: ctrl} + mock.recorder = &MockPayloadWriterMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use +func (m *MockPayloadWriter) EXPECT() *MockPayloadWriterMockRecorder { + return m.recorder +} + // WritePayload mocks base method -func (m *MockPayloadStore) WritePayload(payloadHash hash.SHA256Hash, data []byte) error { +func (m *MockPayloadWriter) WritePayload(payloadHash hash.SHA256Hash, data []byte) error { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "WritePayload", payloadHash, data) ret0, _ := ret[0].(error) @@ -314,7 +375,42 @@ func (m *MockPayloadStore) WritePayload(payloadHash hash.SHA256Hash, data []byte } // WritePayload indicates an expected call of WritePayload -func (mr *MockPayloadStoreMockRecorder) WritePayload(payloadHash, data interface{}) *gomock.Call { +func (mr *MockPayloadWriterMockRecorder) WritePayload(payloadHash, data interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WritePayload", reflect.TypeOf((*MockPayloadStore)(nil).WritePayload), payloadHash, data) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WritePayload", reflect.TypeOf((*MockPayloadWriter)(nil).WritePayload), payloadHash, data) +} + +// MockObservable is a mock of Observable interface +type MockObservable struct { + ctrl *gomock.Controller + recorder *MockObservableMockRecorder +} + +// MockObservableMockRecorder is the mock recorder for MockObservable +type MockObservableMockRecorder struct { + mock *MockObservable +} + +// NewMockObservable creates a new mock instance +func NewMockObservable(ctrl *gomock.Controller) *MockObservable { + mock := &MockObservable{ctrl: ctrl} + mock.recorder = &MockObservableMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use +func (m *MockObservable) EXPECT() *MockObservableMockRecorder { + return m.recorder +} + +// RegisterObserver mocks base method +func (m *MockObservable) RegisterObserver(observer Observer) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "RegisterObserver", observer) +} + +// RegisterObserver indicates an expected call of RegisterObserver +func (mr *MockObservableMockRecorder) RegisterObserver(observer interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterObserver", reflect.TypeOf((*MockObservable)(nil).RegisterObserver), observer) } diff --git a/network/dag/publisher.go b/network/dag/publisher.go new file mode 100644 index 0000000000..e369bf882e --- /dev/null +++ b/network/dag/publisher.go @@ -0,0 +1,89 @@ +package dag + +import ( + "github.com/nuts-foundation/nuts-node/crypto/hash" + "github.com/nuts-foundation/nuts-node/network/log" +) + +// NewReplayingDAGPublisher creates a DAG publisher that replays the complete DAG to all subscribers when started. +func NewReplayingDAGPublisher(payloadStore PayloadStore, dag DAG) Publisher { + publisher := &replayingDAGPublisher{ + subscribers: map[string]Receiver{}, + algo: NewBFSWalkerAlgorithm().(*bfsWalkerAlgorithm), + payloadStore: payloadStore, + dag: dag, + } + dag.RegisterObserver(publisher.DocumentAdded) + payloadStore.RegisterObserver(publisher.PayloadWritten) + return publisher +} + +type replayingDAGPublisher struct { + subscribers map[string]Receiver + algo *bfsWalkerAlgorithm + payloadStore PayloadStore + dag DAG +} + +func (s *replayingDAGPublisher) PayloadWritten(_ interface{}) { + s.publish(hash.EmptyHash()) +} + +func (s *replayingDAGPublisher) DocumentAdded(document interface{}) { + doc := document.(Document) + // Received new document, add it to the subscription walker resume list so it resumes from this document + // when the payload is received. + s.algo.resumeAt.PushBack(doc.Ref()) + s.publish(doc.Ref()) +} + +func (s *replayingDAGPublisher) Subscribe(documentType string, receiver Receiver) { + oldSubscriber := s.subscribers[documentType] + s.subscribers[documentType] = func(document Document, payload []byte) error { + // Chain subscribers in case there's more than 1 + if oldSubscriber != nil { + if err := oldSubscriber(document, payload); err != nil { + return err + } + } + return receiver(document, payload) + } +} + +func (s replayingDAGPublisher) Start() { + root, err := s.dag.Root() + if err != nil { + log.Logger().Errorf("Unable to retrieve DAG root for replaying subscriptions: %v", err) + return + } + if !root.Empty() { + s.publish(root) + } +} + +func (s *replayingDAGPublisher) publish(startAt hash.SHA256Hash) { + err := s.dag.Walk(s.algo, s.publishDocument, startAt) + if err != nil { + log.Logger().Errorf("Unable to publish DAG: %v", err) + } +} + +func (s *replayingDAGPublisher) publishDocument(document Document) bool { + receiver := s.subscribers[document.PayloadType()] + if receiver == nil { + return true + } + payload, err := s.payloadStore.ReadPayload(document.Payload()) + if err != nil { + log.Logger().Errorf("Unable to read payload to publish DAG: (ref=%s) %v", document.Ref(), err) + return false + } + if payload == nil { + // We haven't got the payload, break of processing for this branch + return false + } + if err := receiver(document, payload); err != nil { + log.Logger().Errorf("Document subscriber returned an error (ref=%s,type=%s): %v", document.Ref(), document.PayloadType(), err) + } + return true +} diff --git a/network/dag/publisher_test.go b/network/dag/publisher_test.go new file mode 100644 index 0000000000..148d52215e --- /dev/null +++ b/network/dag/publisher_test.go @@ -0,0 +1,165 @@ +package dag + +import ( + "errors" + "github.com/golang/mock/gomock" + "github.com/nuts-foundation/nuts-node/test/io" + "github.com/stretchr/testify/assert" + "testing" +) + +func TestReplayingPublisher(t *testing.T) { + t.Run("empty graph at start", func(t *testing.T) { + testDirectory := io.TestDirectory(t) + db := createBBoltDB(testDirectory) + dag := NewBBoltDAG(db) + payloadStore := NewBBoltPayloadStore(db) + publisher := NewReplayingDAGPublisher(payloadStore, dag).(*replayingDAGPublisher) + received := false + document := CreateTestDocumentWithJWK(1) + publisher.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { + assert.Equal(t, document, actualDocument) + received = true + return nil + }) + publisher.Start() + + // Now add document and write payload to trigger the observers + dag.Add(document) + payloadStore.WritePayload(document.Payload(), []byte{1, 2, 3}) + + assert.True(t, received) + }) + t.Run("non-empty graph at start", func(t *testing.T) { + testDirectory := io.TestDirectory(t) + db := createBBoltDB(testDirectory) + dag := NewBBoltDAG(db) + payloadStore := NewBBoltPayloadStore(db) + document := CreateTestDocumentWithJWK(1) + err := dag.Add(document) + if !assert.NoError(t, err) { + return + } + err = payloadStore.WritePayload(document.Payload(), []byte{1, 2, 3}) + if !assert.NoError(t, err) { + return + } + + publisher := NewReplayingDAGPublisher(payloadStore, dag).(*replayingDAGPublisher) + received := false + publisher.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { + assert.Equal(t, document, actualDocument) + received = true + return nil + }) + publisher.Start() + + assert.True(t, received) + }) +} + +func TestReplayingPublisher_publishDocument(t *testing.T) { + t.Run("no subscribers", func(t *testing.T) { + publisher, ctrl, _ := createPublisher(t) + defer ctrl.Finish() + + publisher.publishDocument(CreateTestDocumentWithJWK(1)) + }) + t.Run("single subscriber", func(t *testing.T) { + publisher, ctrl, store := createPublisher(t) + defer ctrl.Finish() + + document := CreateTestDocumentWithJWK(1) + store.EXPECT().ReadPayload(document.Payload()).Return([]byte{1, 2, 3}, nil) + + received := false + publisher.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { + assert.Equal(t, document, actualDocument) + received = true + return nil + }) + publisher.publishDocument(document) + assert.True(t, received) + }) + t.Run("payload not present (but present later)", func(t *testing.T) { + publisher, ctrl, store := createPublisher(t) + defer ctrl.Finish() + + document := CreateTestDocumentWithJWK(1) + store.EXPECT().ReadPayload(document.Payload()).Return(nil, nil) + + received := false + publisher.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { + assert.Equal(t, document, actualDocument) + received = true + return nil + }) + publisher.publishDocument(document) + assert.False(t, received) + + // Now add the payload and trigger observer func + store.EXPECT().ReadPayload(document.Payload()).Return([]byte{1, 2, 3}, nil) + publisher.publishDocument(document) + + assert.True(t, received) + }) + t.Run("error reading payload", func(t *testing.T) { + publisher, ctrl, store := createPublisher(t) + defer ctrl.Finish() + + document := CreateTestDocumentWithJWK(1) + store.EXPECT().ReadPayload(document.Payload()).Return(nil, errors.New("failed")) + + received := false + publisher.Subscribe(document.PayloadType(), func(actualDocument Document, actualPayload []byte) error { + received = true + return nil + }) + publisher.publishDocument(document) + assert.False(t, received) + }) + t.Run("multiple subscribers", func(t *testing.T) { + publisher, ctrl, store := createPublisher(t) + defer ctrl.Finish() + + document := CreateTestDocumentWithJWK(1) + store.EXPECT().ReadPayload(document.Payload()).Return([]byte{1, 2, 3}, nil) + + calls := 0 + receiver := func(actualDocument Document, actualPayload []byte) error { + calls++ + return nil + } + publisher.Subscribe(document.PayloadType(), receiver) + publisher.Subscribe(document.PayloadType(), receiver) + + publisher.publishDocument(document) + assert.Equal(t, 2, calls) + }) + t.Run("multiple subscribers, first fails", func(t *testing.T) { + publisher, ctrl, store := createPublisher(t) + defer ctrl.Finish() + + document := CreateTestDocumentWithJWK(1) + store.EXPECT().ReadPayload(document.Payload()).Return([]byte{1, 2, 3}, nil) + calls := 0 + receiver := func(actualDocument Document, actualPayload []byte) error { + calls++ + return errors.New("failed") + } + publisher.Subscribe(document.PayloadType(), receiver) + publisher.Subscribe(document.PayloadType(), receiver) + publisher.publishDocument(document) + assert.Equal(t, 1, calls) + }) +} + +func createPublisher(t *testing.T) (*replayingDAGPublisher, *gomock.Controller, *MockPayloadStore) { + ctrl := gomock.NewController(t) + payloadStore := NewMockPayloadStore(ctrl) + payloadStore.EXPECT().RegisterObserver(gomock.Any()) + dag := NewMockDAG(ctrl) + dag.EXPECT().RegisterObserver(gomock.Any()) + publisher := NewReplayingDAGPublisher(payloadStore, dag).(*replayingDAGPublisher) + return publisher, ctrl, payloadStore +} diff --git a/network/dag/test.go b/network/dag/test.go index 8b4cae6fb7..1552c7bd37 100644 --- a/network/dag/test.go +++ b/network/dag/test.go @@ -54,3 +54,38 @@ func CreateTestDocument(num uint32, prevs ...hash.SHA256Hash) (Document, string, } return signedDocument, kid, signer.Key.Public() } + +// graphF creates the following graph: +//..................A +//................/ \ +//...............B C +//...............\ / \ +//.................D E +//.......................\ +//........................F +func graphF() []Document { + A := CreateTestDocumentWithJWK(0) + B := CreateTestDocumentWithJWK(1, A.Ref()) + C := CreateTestDocumentWithJWK(2, A.Ref()) + D := CreateTestDocumentWithJWK(3, B.Ref(), C.Ref()) + E := CreateTestDocumentWithJWK(4, C.Ref()) + F := CreateTestDocumentWithJWK(5, E.Ref()) + return []Document{A, B, C, D, E, F} +} + +// graphG creates the following graph: +//..................A +//................/ \ +//...............B C +//...............\ / \ +//.................D E +//.................\.....\ +//..................\.....F +//...................\.../ +//.....................G +func graphG() []Document { + docs := graphF() + g := CreateTestDocumentWithJWK(6, docs[3].Ref(), docs[5].Ref()) + docs = append(docs, g) + return docs +} diff --git a/network/network.go b/network/network.go index e711770043..2904f3e925 100644 --- a/network/network.go +++ b/network/network.go @@ -30,10 +30,14 @@ import ( "github.com/nuts-foundation/nuts-node/network/p2p" "github.com/nuts-foundation/nuts-node/network/proto" "github.com/pkg/errors" + "go.etcd.io/bbolt" "sync" "time" ) +// boltDBFileMode holds the Unix file mode the created BBolt database files will have. +const boltDBFileMode = 0600 + // ModuleName defines the name of this module const ModuleName = "Network" @@ -45,6 +49,7 @@ type NetworkEngine struct { p2pNetwork p2p.P2PNetwork protocol proto.Protocol documentGraph dag.DAG + publisher dag.Publisher payloadStore dag.PayloadStore keyStore crypto.KeyStore } @@ -64,9 +69,14 @@ func NewNetworkInstance(config Config, keyStore crypto.KeyStore) *NetworkEngine func (n *NetworkEngine) Configure() error { var err error n.configOnce.Do(func() { - if n.documentGraph, n.payloadStore, err = dag.NewBBoltDAG(n.Config.DatabaseFile); err != nil { + db, bboltErr := bbolt.Open(n.Config.DatabaseFile, boltDBFileMode, bbolt.DefaultOptions) + if bboltErr != nil { + err = fmt.Errorf("unable to create bbolt database: %w", err) return } + n.documentGraph = dag.NewBBoltDAG(db) + n.payloadStore = dag.NewBBoltPayloadStore(db) + n.publisher = dag.NewReplayingDAGPublisher(n.payloadStore, n.documentGraph) peerID := p2p.PeerID(uuid.New().String()) n.protocol.Configure(n.p2pNetwork, n.documentGraph, n.payloadStore, dag.NewDocumentSignatureVerifier(n.keyStore), time.Duration(n.Config.AdvertHashesInterval)*time.Millisecond, peerID) networkConfig, p2pErr := n.buildP2PConfig(peerID) @@ -92,13 +102,14 @@ func (n *NetworkEngine) Start() error { log.Logger().Warn("NetworkEngine is in offline mode (P2P layer not configured).") } n.protocol.Start() + n.publisher.Start() return nil } // Subscribe makes a subscription for the specified document type. The receiver is called when a document // is received for the specified type. func (n *NetworkEngine) Subscribe(documentType string, receiver dag.Receiver) { - n.documentGraph.Subscribe(documentType, receiver) + n.publisher.Subscribe(documentType, receiver) } // GetDocument retrieves the document for the given reference. If the document is not known, an error is returned. diff --git a/network/network_integration_test.go b/network/network_integration_test.go index 451db66d92..85d42c4b5d 100644 --- a/network/network_integration_test.go +++ b/network/network_integration_test.go @@ -122,7 +122,7 @@ func TestNetworkIntegration_SignatureIncorrect(t *testing.T) { // Start node 1 and node 2. Node 1 adds 3 documents: // 1. first document is OK, must be received // 2. second document has an invalid signature, must be rejected - // 3. third document is OK, must be received (to deal with timing issues) + // 3. third document is OK, must be received (to deal with timing issues) node1, err := startNode("node1", path.Join(testDirectory, "node1")) if !assert.NoError(t, err) { return @@ -149,11 +149,12 @@ func TestNetworkIntegration_SignatureIncorrect(t *testing.T) { node1.payloadStore.WritePayload(hash.SHA256Sum(payload), payload) _ = node1.documentGraph.Add(craftedDocument) // Send third OK document - if !addDocumentAndWaitForItToArrive(t, "third document", node1, "node2") { + if !addDocumentAndWaitForItToArrive(t, "third document", node2, "node1") { return } - // Assert node2 only processed the first and last document + // Assert node2 only processed the first and last document, node1 all 3 assert.Len(t, receivedDocuments["node2"], 2) + assert.Len(t, receivedDocuments["node1"], 3) for _, d := range receivedDocuments["node2"] { if d.Ref().Equals(craftedDocument.Ref()) { t.Error("Node 2 processed the crafted document.") @@ -223,7 +224,7 @@ func startNode(name string, directory string) (*NetworkEngine, error) { instance.Subscribe(documentType, func(document dag.Document, payload []byte) error { mutex.Lock() defer mutex.Unlock() - println("document", string(payload), "arrived at", name) + log.Logger().Infof("document %s arrived at %s", string(payload), name) receivedDocuments[name] = append(receivedDocuments[name], document) return nil }) diff --git a/network/network_test.go b/network/network_test.go index 7dd0b9f6c3..478b6cd662 100644 --- a/network/network_test.go +++ b/network/network_test.go @@ -43,6 +43,7 @@ type networkTestContext struct { graph *dag.MockDAG payload *dag.MockPayloadStore keyStore *crypto.MockKeyStore + publisher *dag.MockPublisher } func TestNetwork_ListDocuments(t *testing.T) { @@ -84,7 +85,7 @@ func TestNetwork_Subscribe(t *testing.T) { defer ctrl.Finish() t.Run("ok", func(t *testing.T) { cxt := createNetwork(t, ctrl) - cxt.graph.EXPECT().Subscribe("some-type", nil) + cxt.publisher.EXPECT().Subscribe("some-type", nil) cxt.network.Subscribe("some-type", nil) }) } @@ -132,6 +133,7 @@ func TestNetwork_CreateDocument(t *testing.T) { cxt.keyStore.EXPECT().SignJWS(gomock.Any(), gomock.Any(), gomock.Eq("signing-key")).DoAndReturn(func(payload []byte, protectedHeaders map[string]interface{}, kid interface{}) (string, error) { return crypto.NewTestSigner().SignJWS(payload, protectedHeaders, "") }) + cxt.publisher.EXPECT().Start() err := cxt.network.Start() if !assert.NoError(t, err) { return @@ -153,6 +155,7 @@ func TestNetwork_CreateDocument(t *testing.T) { cxt.keyStore.EXPECT().SignJWS(gomock.Any(), gomock.Any(), gomock.Eq("signing-key")).DoAndReturn(func(payload []byte, protectedHeaders map[string]interface{}, kid interface{}) (string, error) { return crypto.NewTestSigner().SignJWS(payload, protectedHeaders, "") }) + cxt.publisher.EXPECT().Start() err := cxt.network.Start() if !assert.NoError(t, err) { return @@ -170,6 +173,7 @@ func TestNetwork_Start(t *testing.T) { cxt.p2pNetwork.EXPECT().Start() cxt.p2pNetwork.EXPECT().Configured().Return(true) cxt.protocol.EXPECT().Start() + cxt.publisher.EXPECT().Start() err := cxt.network.Start() if !assert.NoError(t, err) { return @@ -179,6 +183,7 @@ func TestNetwork_Start(t *testing.T) { cxt := createNetwork(t, ctrl) cxt.p2pNetwork.EXPECT().Configured().Return(false) cxt.protocol.EXPECT().Start() + cxt.publisher.EXPECT().Start() err := cxt.network.Start() if !assert.NoError(t, err) { return @@ -254,6 +259,7 @@ func createNetwork(t *testing.T, ctrl *gomock.Controller) *networkTestContext { protocol := proto.NewMockProtocol(ctrl) graph := dag.NewMockDAG(ctrl) payload := dag.NewMockPayloadStore(ctrl) + publisher := dag.NewMockPublisher(ctrl) testDirectory := io.TestDirectory(t) networkConfig := TestNetworkConfig(testDirectory) networkConfig.TrustStoreFile = "test/truststore.pem" @@ -266,12 +272,14 @@ func createNetwork(t *testing.T, ctrl *gomock.Controller) *networkTestContext { network.protocol = protocol network.documentGraph = graph network.payloadStore = payload + network.publisher = publisher return &networkTestContext{ network: network, p2pNetwork: p2pNetwork, protocol: protocol, graph: graph, payload: payload, + publisher: publisher, keyStore: keyStore, } } diff --git a/network/proto/handlers.go b/network/proto/handlers.go index 2c8af46132..1fc4947d35 100644 --- a/network/proto/handlers.go +++ b/network/proto/handlers.go @@ -84,7 +84,7 @@ func (p *protocol) handleDocumentPayload(peer p2p.PeerID, contents *transport.Do log.Logger().Errorf("Error while looking up document to write payload (payloadHash=%s): %v", payloadHash, err) } else if document == nil { log.Logger().Warnf("Received document payload for document we don't have (payloadHash=%s)", payloadHash) - } else if hasPayload, err := p.payloadStore.IsPayloadPresent(payloadHash); err != nil { + } else if hasPayload, err := p.payloadStore.IsPresent(payloadHash); err != nil { log.Logger().Errorf("Error while checking whether we already have payload (payloadHash=%s): %v", payloadHash, err) } else if hasPayload { log.Logger().Debugf("Received payload we already have (payloadHash=%s)", payloadHash) @@ -151,7 +151,7 @@ func (p *protocol) checkDocumentOnLocalNode(peer p2p.PeerID, documentRef hash.SH return fmt.Errorf("unable to add received document to DAG: %w", err) } queryContents = true - } else if payloadPresent, err := p.payloadStore.IsPayloadPresent(document.Payload()); err != nil { + } else if payloadPresent, err := p.payloadStore.IsPresent(document.Payload()); err != nil { return err } else { queryContents = !payloadPresent