Improve nops broadcasting. (#150)

* ledger: add failing test

* ledger: stop nop broadcast if depth > last added tx depth

* ledger: wait for faucet in test

* ledger: fix panic due to closed channel

* ledger: check if nop broadcast is stopped in test

* ledger: increase consensus timeout in test

* ledger: rename test and add comments

* ledger: disable GC when running tests

* ledger, snowball: revert test code

* ledger: poll round to wait for consensus

* ledger: don't print missing parent error

* ledger: allow test to run up to 5 minutes

* api, cmd/graph, cmd/wavelet, ledger: use options

* ledger: remove stray defer statement

* ledger: reduce number of nodes and add sleep

* ledger: add failing test to check excessive consensus round

* ledger: fix excessive nop

* ledger: revert error silencing

* ledger: improve test

Author: Hasyimi Bahrudin

accounts.go

@@ -21,13 +21,14 @@ package wavelet
 
 import (
 	"context"
-	"github.com/perlin-network/wavelet/avl"
-	"github.com/perlin-network/wavelet/store"
-	"github.com/pkg/errors"
 	"sync"
 	"sync/atomic"
 	"time"
 	"unsafe"
+
+	"github.com/perlin-network/wavelet/avl"
+	"github.com/perlin-network/wavelet/store"
+	"github.com/pkg/errors"
 )
 
 type Accounts struct {

api/mod_test.go

@@ -26,15 +26,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"github.com/buaazp/fasthttprouter"
-	"github.com/perlin-network/noise/skademlia"
-	"github.com/perlin-network/wavelet"
-	"github.com/perlin-network/wavelet/store"
-	"github.com/perlin-network/wavelet/sys"
-	"github.com/pkg/errors"
-	"github.com/stretchr/testify/assert"
-	"github.com/valyala/fasthttp"
-	"github.com/valyala/fastjson"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -44,6 +35,16 @@ import (
 	"testing"
 	"testing/quick"
 	"time"
+
+	"github.com/buaazp/fasthttprouter"
+	"github.com/perlin-network/noise/skademlia"
+	"github.com/perlin-network/wavelet"
+	"github.com/perlin-network/wavelet/store"
+	"github.com/perlin-network/wavelet/sys"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/valyala/fasthttp"
+	"github.com/valyala/fastjson"
 )
 
 func TestListTransaction(t *testing.T) {
@@ -744,7 +745,7 @@ func createLedger(t *testing.T) *wavelet.Ledger {
 	keys, err := skademlia.NewKeys(1, 1)
 	assert.NoError(t, err)
 
-	ledger := wavelet.NewLedger(store.NewInmem(), skademlia.NewClient(":0", keys), nil)
+	ledger := wavelet.NewLedger(store.NewInmem(), skademlia.NewClient(":0", keys))
 	return ledger
 }
 

api/mod_ws_test.go

@@ -44,7 +44,7 @@ func TestPollLog(t *testing.T) {
 	keys, err := skademlia.NewKeys(1, 1)
 	assert.NoError(t, err)
 
-	ledger := wavelet.NewLedger(store.NewInmem(), skademlia.NewClient(":0", keys), nil)
+	ledger := wavelet.NewLedger(store.NewInmem(), skademlia.NewClient(":0", keys))
 
 	go gateway.StartHTTP(8080, nil, ledger, keys)
 	defer gateway.Shutdown()

cmd/graph/main.go

@@ -113,7 +113,7 @@ func main() {
 
 	client.SetCredentials(noise.NewCredentials(addr, handshake.NewECDH(), cipher.NewAEAD(), client.Protocol()))
 
-	ledger := wavelet.NewLedger(store.NewInmem(), client, nil)
+	ledger := wavelet.NewLedger(store.NewInmem(), client)
 
 	go func() {
 		server := client.Listen()

cmd/wavelet/main.go

@@ -23,6 +23,14 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"os"
+	"sort"
+	"strconv"
+	"time"
+
 	"github.com/perlin-network/noise"
 	"github.com/perlin-network/noise/cipher"
 	"github.com/perlin-network/noise/edwards25519"
@@ -38,13 +46,6 @@ import (
 	"google.golang.org/grpc"
 	"gopkg.in/urfave/cli.v1"
 	"gopkg.in/urfave/cli.v1/altsrc"
-	"io/ioutil"
-	"net"
-	"net/http"
-	"os"
-	"sort"
-	"strconv"
-	"time"
 )
 
 import _ "net/http/pprof"
@@ -304,7 +305,7 @@ func start(cfg *Config) {
 		logger.Fatal().Err(err).Msgf("Failed to create/open database located at %q.", cfg.Database)
 	}
 
-	ledger := wavelet.NewLedger(kv, client, cfg.Genesis)
+	ledger := wavelet.NewLedger(kv, client, wavelet.WithGenesis(cfg.Genesis))
 
 	go func() {
 		server := client.Listen()

ledger.go

@@ -25,6 +25,11 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
+	"math/rand"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/perlin-network/noise"
 	"github.com/perlin-network/noise/skademlia"
 	"github.com/perlin-network/wavelet/avl"
@@ -36,10 +41,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/peer"
-	"math/rand"
-	"strings"
-	"sync"
-	"time"
 )
 
 type Ledger struct {
@@ -57,9 +58,10 @@ type Ledger struct {
 
 	consensus sync.WaitGroup
 
-	broadcastNops      bool
-	broadcastNopsDelay time.Time
-	broadcastNopsLock  sync.Mutex
+	broadcastNops         bool
+	broadcastNopsMaxDepth uint64
+	broadcastNopsDelay    time.Time
+	broadcastNopsLock     sync.Mutex
 
 	sync      chan struct{}
 	syncVotes chan vote
@@ -70,21 +72,49 @@ type Ledger struct {
 	sendQuota chan struct{}
 }
 
-func NewLedger(kv store.KV, client *skademlia.Client, genesis *string) *Ledger {
+type config struct {
+	GCDisabled bool
+	Genesis    *string
+}
+
+type Option func(cfg *config)
+
+// WithoutGC disables GC. Used for testing purposes.
+func WithoutGC() Option {
+	return func(cfg *config) {
+		cfg.GCDisabled = true
+	}
+}
+
+func WithGenesis(genesis *string) Option {
+	return func(cfg *config) {
+		cfg.Genesis = genesis
+	}
+}
+
+func NewLedger(kv store.KV, client *skademlia.Client, opts ...Option) *Ledger {
+	var cfg config
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+
 	logger := log.Node()
 
 	metrics := NewMetrics(context.TODO())
 	indexer := NewIndexer()
 
 	accounts := NewAccounts(kv)
-	go accounts.GC(context.Background())
+
+	if !cfg.GCDisabled {
+		go accounts.GC(context.Background())
+	}
 
 	rounds, err := NewRounds(kv, sys.PruningLimit)
 
 	var round *Round
 
 	if rounds != nil && err != nil {
-		genesis := performInception(accounts.tree, genesis)
+		genesis := performInception(accounts.tree, cfg.Genesis)
 		if err := accounts.Commit(nil); err != nil {
 			logger.Fatal().Err(err).Msg("BUG: accounts.Commit")
 		}
@@ -162,12 +192,13 @@ func (l *Ledger) AddTransaction(tx Transaction) error {
 		l.gossiper.Push(tx)
 
 		l.broadcastNopsLock.Lock()
-		if tx.Tag != sys.TagNop {
+		if tx.Tag != sys.TagNop && tx.Sender == l.client.Keys().PublicKey() {
+			l.broadcastNops = true
 			l.broadcastNopsDelay = time.Now()
-		}
 
-		if tx.Sender == l.client.Keys().PublicKey() && l.finalizer.Preferred() == nil {
-			l.broadcastNops = true
+			if tx.Depth > l.broadcastNopsMaxDepth {
+				l.broadcastNopsMaxDepth = tx.Depth
+			}
 		}
 		l.broadcastNopsLock.Unlock()
 	}
@@ -271,6 +302,16 @@ func (l *Ledger) Snapshot() *avl.Tree {
 	return l.accounts.Snapshot()
 }
 
+// BroadcastingNop returns true if the node is
+// supposed to broadcast nop transaction.
+func (l *Ledger) BroadcastingNop() bool {
+	l.broadcastNopsLock.Lock()
+	broadcastNops := l.broadcastNops
+	l.broadcastNopsLock.Unlock()
+
+	return broadcastNops
+}
+
 // BroadcastNop has the node send a nop transaction should they have sufficient
 // balance available. They are broadcasted if no other transaction that is not a nop transaction
 // is not broadcasted by the node after 500 milliseconds. These conditions only apply so long as
@@ -387,7 +428,7 @@ func (l *Ledger) PullMissingTransactions() {
 		for _, buf := range batch.Transactions {
 			tx, err := UnmarshalTransaction(bytes.NewReader(buf))
 			if err != nil {
-				fmt.Printf("error unmarshaling downloaded tx [%v]: %+v", err, tx)
+				fmt.Printf("error unmarshaling downloaded tx [%v]: %+v\n", err, tx)
 				continue
 			}
 
@@ -474,8 +515,12 @@ FINALIZE_ROUNDS:
 			continue FINALIZE_ROUNDS
 		}
 
+		// Only stop broadcasting nops if the most recently added transaction
+		// has been applied
 		l.broadcastNopsLock.Lock()
-		l.broadcastNops = false
+		if l.broadcastNops && l.broadcastNopsMaxDepth <= l.finalizer.Preferred().End.Depth {
+			l.broadcastNops = false
+		}
 		l.broadcastNopsLock.Unlock()
 
 		workerChan := make(chan *grpc.ClientConn, 16)

ledger_test.go

@@ -0,0 +1,126 @@
+package wavelet
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestLedger_BroadcastNop checks that:
+//
+// * The ledger will keep broadcasting nop tx as long
+//   as there are unapplied tx (latestTxDepth <= rootDepth).
+//
+// * The ledger will stop broadcasting nop once there
+//   are no more unapplied tx.
+func TestLedger_BroadcastNop(t *testing.T) {
+	testnet := NewTestNetwork(t)
+	defer testnet.Cleanup()
+
+	for i := 0; i < 3; i++ {
+		testnet.AddNode(t, 0)
+	}
+
+	alice := testnet.AddNode(t, 1000000)
+	bob := testnet.AddNode(t, 0)
+
+	// Wait for alice to receive her PERL from the faucet
+	for <-alice.WaitForConsensus() {
+		if alice.Balance() > 0 {
+			break
+		}
+	}
+
+	// Add lots of transactions
+	var txsLock sync.Mutex
+	txs := make([]Transaction, 0, 10000)
+
+	go func() {
+		for i := 0; i < cap(txs); i++ {
+			tx, err := alice.Pay(bob, 1)
+			assert.NoError(t, err)
+
+			txsLock.Lock()
+			txs = append(txs, tx)
+			txsLock.Unlock()
+
+			// Somehow this prevents AddTransaction from
+			// returning ErrMissingParents
+			time.Sleep(time.Nanosecond * 1)
+		}
+	}()
+
+	prevRound := alice.ledger.Rounds().Latest().Index
+	timeout := time.NewTimer(time.Minute * 5)
+	for {
+		select {
+		case <-timeout.C:
+			t.Fatal("timed out before all transactions are applied")
+
+		case <-alice.WaitForConsensus():
+			var appliedCount int
+			var txsCount int
+
+			txsLock.Lock()
+			for _, tx := range txs {
+				if alice.Applied(tx) {
+					appliedCount++
+				}
+				txsCount++
+			}
+			txsLock.Unlock()
+
+			currRound := alice.ledger.Rounds().Latest().Index
+
+			fmt.Printf("%d/%d tx applied, round=%d, root depth=%d\n",
+				appliedCount, txsCount,
+				currRound,
+				alice.ledger.Graph().RootDepth())
+
+			if currRound-prevRound > 1 {
+				t.Fatal("more than 1 round finalized")
+			}
+
+			prevRound = currRound
+
+			if appliedCount < cap(txs) {
+				assert.True(t, alice.ledger.BroadcastingNop(),
+					"node should not stop broadcasting nop while there are unapplied tx")
+			}
+
+			// The test is successful if all tx are applied,
+			// and nop broadcasting is stopped once all tx are applied
+			if appliedCount == cap(txs) && !alice.ledger.BroadcastingNop() {
+				return
+			}
+		}
+	}
+}
+
+func TestLedger_AddTransaction(t *testing.T) {
+	testnet := NewTestNetwork(t)
+	defer testnet.Cleanup()
+
+	alice := testnet.AddNode(t, 0) // alice
+	testnet.AddNode(t, 0)          // bob
+
+	start := alice.ledger.Rounds().Latest().Index
+
+	// Add just 1 transaction
+	_, err := testnet.faucet.PlaceStake(100)
+	assert.NoError(t, err)
+
+	// Try to wait for 2 rounds of consensus.
+	// The second call should result in timeout, because
+	// only 1 round should be finalized.
+	<-alice.WaitForConsensus()
+	<-alice.WaitForConsensus()
+
+	current := alice.ledger.Rounds().Latest().Index
+	if current-start > 1 {
+		t.Fatal("more than 1 round finalized")
+	}
+}