diff --git a/cmd/network.db b/cmd/network.db
new file mode 100644
index 0000000000..e449c28987
Binary files /dev/null and b/cmd/network.db differ
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 <https://www.gnu.org/licenses/>.
- *
- */
-
-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 <https://www.gnu.org/licenses/>.
- *
- */
-
-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