add xor tree repair loop

network/dag/consistency.go

@@ -0,0 +1,160 @@
+/*
+ * Copyright (C) 2023 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 (
+	"context"
+	"github.com/nuts-foundation/go-stoabs"
+	"github.com/nuts-foundation/nuts-node/network/dag/tree"
+	"github.com/nuts-foundation/nuts-node/network/log"
+	"sync"
+	"time"
+)
+
+type circuit int
+
+const (
+	circuitGreen circuit = iota
+	circuitYellow
+	circuitRed
+)
+
+// xorTreeRepair is responsible for repairing the XOR tree. Its loop is triggered when the network layer detects differences in XOR values with all other nodes.
+// It will loop over all pages of the XOR tree and recalculates the XOR value with the transactions in the database.
+// This repair is needed because current networks have nodes that have a wrong XOR value. How this happens is not yet known, it could be due to DB failures of due to bugs in older versions.
+// The fact is that we can fix the state relatively easy.
+// The loop checks a page (512 LC values) per 10 seconds and continues looping until the network layer signals all is ok again.
+type xorTreeRepair struct {
+	ctx          context.Context
+	cancel       context.CancelFunc
+	ticker       *time.Ticker
+	currentPage  uint32
+	state        *state
+	circuitState circuit
+	mutex        sync.Mutex
+}
+
+func newXorTreeRepair(state *state) *xorTreeRepair {
+	return &xorTreeRepair{
+		state:  state,
+		ticker: time.NewTicker(10 * time.Second),
+	}
+}
+
+func (f *xorTreeRepair) start() {
+	var ctx context.Context
+	ctx, f.cancel = context.WithCancel(context.Background())
+	go func() {
+		defer f.ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-f.ticker.C:
+				f.checkPage()
+			}
+		}
+	}()
+}
+
+func (f *xorTreeRepair) shutdown() {
+	if f.cancel != nil {
+		f.cancel()
+	}
+}
+
+func (f *xorTreeRepair) checkPage() {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	// ignore run if circuit is not red
+	if f.circuitState < circuitRed {
+		return
+	}
+
+	currentLC := f.state.lamportClockHigh.Load()
+	lcStart := f.currentPage * PageSize
+	lcEnd := lcStart + PageSize
+
+	// initialize an XOR tree
+	calculatedXorTree := tree.New(tree.NewXor(), PageSize)
+
+	// acquire global lock
+	err := f.state.graph.db.Write(context.Background(), func(txn stoabs.WriteTx) error {
+		txs, err := f.state.graph.findBetweenLC(txn, lcStart, lcEnd)
+		if err != nil {
+			return err
+		}
+		for _, tx := range txs {
+			calculatedXorTree.Insert(tx.Ref(), tx.Clock())
+		}
+
+		// Get XOR leaf from current XOR tree
+		xorTillEnd, _ := f.state.xorTree.getZeroTo(lcEnd - 1)
+		if lcStart != 0 {
+			xorTillStart, _ := f.state.xorTree.getZeroTo(lcStart - 1)
+			_ = xorTillEnd.Subtract(xorTillStart)
+		}
+
+		// Subtract the calculated tree, should be empty if the trees are equal
+		_ = xorTillEnd.Subtract(calculatedXorTree.Root())
+		if !xorTillEnd.Empty() {
+			// it's not empty, so replace the leaf in the current XOR tree with the calculated one
+			err = f.state.xorTree.tree.Replace(lcStart, calculatedXorTree.Root())
+			if err != nil {
+				return err
+			}
+			log.Logger().Warnf("detected XOR tree mismatch for page %d, fixed using recalculated values", f.currentPage)
+		}
+
+		// Now we do the same for the IBLT tree as stated in
+		// https://github.com/nuts-foundation/nuts-node/issues/2295
+		// we skip the iblt tree for now, since the chance for it to become corrupt is incredibly low.
+		// there can only be a problem with duplicate entries, not with missing entries.
+		// the xor tree already has an effect when it's missing entries.
+		// fixing the iblt tree is a copy of the code above (but with ibltTree instead of xorTree).
+
+		return nil
+	})
+	if err != nil {
+		log.Logger().WithError(err).Warnf("Failed to run xorTreeRepair check.")
+	}
+
+	if lcEnd > currentLC {
+		// start over when end is reached for next run
+		f.currentPage = 0
+	} else {
+		// increment page so on the next round we check a different page.
+		f.currentPage++
+	}
+}
+
+func (f *xorTreeRepair) stateOK() {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	f.circuitState = circuitGreen
+}
+
+func (f *xorTreeRepair) incrementCount() {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	f.circuitState++
+}

network/dag/consistency_test.go

@@ -0,0 +1,152 @@
+/*
+ * Copyright (C) 2023 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 (
+	"context"
+	"encoding/binary"
+	"github.com/magiconair/properties/assert"
+	"github.com/nuts-foundation/go-stoabs"
+	"github.com/nuts-foundation/go-stoabs/bbolt"
+	"github.com/nuts-foundation/nuts-node/crypto/hash"
+	"github.com/nuts-foundation/nuts-node/network/dag/tree"
+	"github.com/nuts-foundation/nuts-node/test"
+	"github.com/nuts-foundation/nuts-node/test/io"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestXorTreeRepair(t *testing.T) {
+	t.Cleanup(func() {
+		goleak.VerifyNone(t)
+	})
+
+	tx, _, _ := CreateTestTransaction(1)
+	t.Run("xor tree repaired after 2 signals", func(t *testing.T) {
+		txState := createXorTreeRepairState(t, tx)
+		require.NoError(t, txState.Start())
+		txState.xorTree = newTreeStore(xorShelf, tree.New(tree.NewXor(), PageSize))
+
+		// twice to set circuit to red
+		txState.IncorrectStateDetected()
+		txState.IncorrectStateDetected()
+
+		// await for XOR to change
+		test.WaitFor(t, func() (bool, error) {
+			txState.xorTreeRepair.mutex.Lock()
+			defer txState.xorTreeRepair.mutex.Unlock()
+
+			xorRoot := txState.xorTree.tree.Root()
+			hashRoot := xorRoot.(*tree.Xor).Hash()
+			return hashRoot.Equals(tx.Ref()), nil
+		}, time.Second, "xorTree not updated within wait period")
+	})
+	t.Run("checkPage executed after 2 signals", func(t *testing.T) {
+		txState := createXorTreeRepairState(t, tx)
+		txState.xorTree = newTreeStore(xorShelf, tree.New(tree.NewXor(), PageSize))
+
+		// twice to set circuit to red
+		txState.IncorrectStateDetected()
+		txState.IncorrectStateDetected()
+		txState.xorTreeRepair.checkPage()
+
+		assert.Equal(t, tx.Ref(), xorRootDate(txState))
+	})
+	t.Run("checkPage not executed after 1 signal", func(t *testing.T) {
+		txState := createXorTreeRepairState(t, tx)
+		txState.xorTree = newTreeStore(xorShelf, tree.New(tree.NewXor(), PageSize))
+
+		// twice to set circuit to yellow
+		txState.IncorrectStateDetected()
+		txState.xorTreeRepair.checkPage()
+
+		assert.Equal(t, hash.EmptyHash(), xorRootDate(txState))
+	})
+	t.Run("checkPage not executed after okState", func(t *testing.T) {
+		txState := createXorTreeRepairState(t, tx)
+		txState.xorTree = newTreeStore(xorShelf, tree.New(tree.NewXor(), PageSize))
+
+		// twice to set circuit to red
+		txState.IncorrectStateDetected()
+		txState.IncorrectStateDetected()
+		// back to green
+		txState.CorrectStateDetected()
+		txState.xorTreeRepair.checkPage()
+
+		assert.Equal(t, hash.EmptyHash(), xorRootDate(txState))
+	})
+	t.Run("checkPage executed for multiple pages", func(t *testing.T) {
+		txState := createXorTreeRepairState(t, tx)
+		prev := tx
+		expectedHash := tx.Ref()
+		for i := uint32(1); i < 600; i++ {
+			tx2, _, _ := CreateTestTransaction(i, prev)
+			payload := make([]byte, 4)
+			binary.BigEndian.PutUint32(payload, i)
+			_ = txState.Add(context.Background(), tx2, payload)
+			prev = tx2
+			expectedHash = expectedHash.Xor(tx2.Ref())
+		}
+		require.NoError(t, txState.Start())
+		txState.xorTree = newTreeStore(xorShelf, tree.New(tree.NewXor(), PageSize))
+
+		// twice to set circuit to red
+		txState.IncorrectStateDetected()
+		txState.IncorrectStateDetected()
+
+		// await for XOR to change
+		test.WaitFor(t, func() (bool, error) {
+			txState.xorTreeRepair.mutex.Lock()
+			defer txState.xorTreeRepair.mutex.Unlock()
+
+			xorRoot := txState.xorTree.tree.Root()
+			hashRoot := xorRoot.(*tree.Xor).Hash()
+			return hashRoot.Equals(expectedHash), nil
+		}, 5*time.Second, "xorTree not updated within wait period")
+	})
+}
+
+func xorRootDate(s *state) hash.SHA256Hash {
+	return s.xorTree.tree.Root().(*tree.Xor).Hash()
+}
+
+func createXorTreeRepairState(t testing.TB, tx Transaction) *state {
+	txState := createStoppedState(t)
+	txState.xorTreeRepair.ticker = time.NewTicker(5 * time.Millisecond)
+	payload := []byte{0, 0, 0, 1}
+	txState.Add(context.Background(), tx, payload)
+	return txState
+}
+
+func createStoppedState(t testing.TB) *state {
+	testDir := io.TestDirectory(t)
+	bboltStore, err := bbolt.CreateBBoltStore(filepath.Join(testDir, "test_state"), stoabs.WithNoSync())
+	if err != nil {
+		t.Fatal("failed to create store: ", err)
+	}
+	s, err := NewState(bboltStore)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		s.Shutdown()
+	})
+	return s.(*state)
+}

network/dag/interface.go

@@ -93,6 +93,11 @@ type State interface {
 	//	- highest lamport clock in the DAG
 	// A requested clock of math.MaxUint32 will return the iblt of the entire DAG
 	IBLT(reqClock uint32) (tree.Iblt, uint32)
+
+	// IncorrectStateDetected is called when the xor and LC value from a gossip message do NOT match the local state.
+	IncorrectStateDetected()
+	// CorrectStateDetected is called when the xor and LC value from a gossip message match the local state.
+	CorrectStateDetected()
 }
 
 // Statistics holds data about the current state of the DAG.