meshdb transaction dedupe

cmd/node/node.go

@@ -280,7 +280,8 @@ func (app *SpacemeshApp) initServices(instanceName string, swarm server.Service,
 	ld := time.Duration(app.Config.LayerDurationSec) * time.Second
 	clock := timesync.NewTicker(timesync.RealClock{}, ld, gTime)
 	trtl := consensus.NewAlgorithm(consensus.NewNinjaTortoise(layerSize, lg.WithName("trtl")))
-	msh := mesh.NewMesh(db, db, db, app.Config.REWARD, trtl, processor, lg.WithName("mesh")) //todo: what to do with the logger?
+	mdb := mesh.NewMeshDB(db, db, db, db, lg.WithName("meshDb"))
+	msh := mesh.NewMesh(mdb, app.Config.REWARD, trtl, processor, lg.WithName("mesh")) //todo: what to do with the logger?
 
 	conf := sync.Configuration{SyncInterval: 1 * time.Second, Concurrency: 4, LayerSize: int(layerSize), RequestTimeout: 100 * time.Millisecond}
 	syncer := sync.NewSync(swarm, msh, blockOracle, conf, clock.Subscribe(), lg)

mesh/block.go

@@ -6,14 +6,25 @@ import (
 	"github.com/davecgh/go-xdr/xdr2"
 	"github.com/google/uuid"
 	"github.com/spacemeshos/go-spacemesh/address"
-	"io"
 	"math/big"
 	"time"
 )
 
 type BlockID uint64
-
 type LayerID uint64
+type TransactionId []byte
+
+type BlockHeader struct {
+	Id         BlockID
+	LayerIndex LayerID
+	MinerID    string
+	Data       []byte
+	Coin       bool
+	Timestamp  int64
+	BlockVotes []BlockID
+	ViewEdges  []BlockID
+	TxIds      []TransactionId
+}
 
 type Block struct {
 	Id         BlockID
@@ -22,17 +33,17 @@ type Block struct {
 	Data       []byte
 	Coin       bool
 	Timestamp  int64
-	Txs        []SerializableTransaction
+	Txs        []*SerializableTransaction
 	BlockVotes []BlockID
 	ViewEdges  []BlockID
 }
 
 type SerializableTransaction struct {
 	AccountNonce uint64
-	Price        []byte
-	GasLimit     uint64
 	Recipient    *address.Address
 	Origin       address.Address //todo: remove this, should be calculated from sig.
+	GasLimit     uint64
+	Price        []byte
 	Amount       []byte
 	Payload      []byte
 }
@@ -55,14 +66,35 @@ func NewBlock(coin bool, data []byte, ts time.Time, LayerID LayerID) *Block {
 		LayerIndex: LayerID,
 		BlockVotes: make([]BlockID, 0, 10),
 		ViewEdges:  make([]BlockID, 0, 10),
-		Txs:        make([]SerializableTransaction, 0, 10),
+		Txs:        make([]*SerializableTransaction, 0, 10),
 		Timestamp:  ts.UnixNano(),
 		Data:       data,
 		Coin:       coin,
 	}
 	return &b
 }
 
+func newBlockHeader(block *Block) *BlockHeader {
+	tids := make([]TransactionId, len(block.Txs))
+	for idx, tx := range block.Txs {
+		//keep tx's in same order !!!
+		tids[idx] = getTransactionId(tx)
+	}
+
+	b := BlockHeader{
+		Id:         block.ID(),
+		LayerIndex: block.Layer(),
+		BlockVotes: block.BlockVotes,
+		ViewEdges:  block.ViewEdges,
+		Timestamp:  block.Timestamp,
+		Data:       block.Data,
+		Coin:       block.Coin,
+		MinerID:    block.MinerID,
+		TxIds:      tids,
+	}
+	return &b
+}
+
 func NewSerializableTransaction(nonce uint64, origin, recepient address.Address, amount, price *big.Int, gasLimit uint64) *SerializableTransaction {
 	return &SerializableTransaction{
 		AccountNonce: nonce,
@@ -94,7 +126,7 @@ func (b *Block) AddView(id BlockID) {
 }
 
 func (b *Block) AddTransaction(sr *SerializableTransaction) {
-	b.Txs = append(b.Txs, *sr)
+	b.Txs = append(b.Txs, sr)
 }
 
 type Layer struct {
@@ -131,6 +163,14 @@ func NewExistingLayer(idx LayerID, blocks []*Block) *Layer {
 	return &l
 }
 
+func getBlockHeaderBytes(bheader *BlockHeader) ([]byte, error) {
+	var w bytes.Buffer
+	if _, err := xdr.Marshal(&w, bheader); err != nil {
+		return nil, fmt.Errorf("error marshalling block ids %v", err)
+	}
+	return w.Bytes(), nil
+}
+
 func BlockAsBytes(block Block) ([]byte, error) {
 	var w bytes.Buffer
 	if _, err := xdr.Marshal(&w, &block); err != nil {
@@ -139,9 +179,18 @@ func BlockAsBytes(block Block) ([]byte, error) {
 	return w.Bytes(), nil
 }
 
-func BytesAsBlock(buf io.Reader) (Block, error) {
+func BytesAsBlockHeader(buf []byte) (BlockHeader, error) {
+	b := BlockHeader{}
+	_, err := xdr.Unmarshal(bytes.NewReader(buf), &b)
+	if err != nil {
+		return b, err
+	}
+	return b, nil
+}
+
+func BytesAsBlock(buf []byte) (Block, error) {
 	b := Block{}
-	_, err := xdr.Unmarshal(buf, &b)
+	_, err := xdr.Unmarshal(bytes.NewReader(buf), &b)
 	if err != nil {
 		return b, err
 	}
@@ -156,9 +205,9 @@ func TransactionAsBytes(tx *SerializableTransaction) ([]byte, error) {
 	return w.Bytes(), nil
 }
 
-func BytesAsTransaction(buf io.Reader) (*SerializableTransaction, error) {
+func BytesAsTransaction(buf []byte) (*SerializableTransaction, error) {
 	b := SerializableTransaction{}
-	_, err := xdr.Unmarshal(buf, &b)
+	_, err := xdr.Unmarshal(bytes.NewReader(buf), &b)
 	if err != nil {
 		return &b, err
 	}

mesh/mesh.go

@@ -6,7 +6,6 @@ import (
 	"github.com/spacemeshos/go-spacemesh/address"
 	"github.com/spacemeshos/go-spacemesh/common"
 	"github.com/spacemeshos/go-spacemesh/crypto/sha3"
-	"github.com/spacemeshos/go-spacemesh/database"
 	"github.com/spacemeshos/go-spacemesh/log"
 	"github.com/spacemeshos/go-spacemesh/rlp"
 	"math/big"
@@ -48,14 +47,14 @@ type Mesh struct {
 	done          chan struct{}
 }
 
-func NewMesh(layers, blocks, validity database.DB, rewardConfig RewardConfig, mesh MeshValidator, state StateUpdater, logger log.Log) *Mesh {
+func NewMesh(mDb *meshDB, rewardConfig RewardConfig, mesh MeshValidator, state StateUpdater, logger log.Log) *Mesh {
 	//todo add boot from disk
 	ll := &Mesh{
 		Log:          logger,
 		tortoise:     mesh,
 		state:        state,
 		done:         make(chan struct{}),
-		meshDB:       NewMeshDB(layers, blocks, validity, logger),
+		meshDB:       mDb,
 		rewardConfig: rewardConfig,
 	}
 
@@ -176,7 +175,7 @@ func (m *Mesh) ExtractUniqueOrderedTransactions(l *Layer) []*Transaction {
 		}
 		for _, tx := range b.Txs {
 			//todo: think about these conversions.. are they needed?
-			txs = append(txs, SerializableTransaction2StateTransaction(&tx))
+			txs = append(txs, SerializableTransaction2StateTransaction(tx))
 		}
 	}
 
@@ -238,7 +237,7 @@ func (m *Mesh) handleOrphanBlocks(block *Block) {
 		m.orphanBlocks[block.Layer()] = make(map[BlockID]struct{})
 	}
 	m.orphanBlocks[block.Layer()][block.ID()] = struct{}{}
-	m.Info("Added block %d to orphans", block.ID())
+	m.Debug("Added block %d to orphans", block.ID())
 	atomic.AddInt32(&m.orphanBlockCount, 1)
 	for _, b := range block.ViewEdges {
 		for _, layermap := range m.orphanBlocks {

mesh/mesh_test.go

@@ -2,6 +2,7 @@ package mesh
 
 import (
 	"bytes"
+	"fmt"
 	"github.com/spacemeshos/go-spacemesh/database"
 	"github.com/spacemeshos/go-spacemesh/log"
 	"github.com/stretchr/testify/assert"
@@ -31,10 +32,10 @@ func (MockState) ApplyRewards(layer LayerID, miners map[string]struct{}, underQu
 func getMesh(id string) *Mesh {
 
 	//time := time.Now()
-	bdb := database.NewMemDatabase()
-	ldb := database.NewMemDatabase()
-	cdb := database.NewMemDatabase()
-	layers := NewMesh(ldb, bdb, cdb, ConfigTst(), &MeshValidatorMock{}, &MockState{}, log.New(id, "", ""))
+	db := database.NewMemDatabase()
+	lg := log.New(id, "", "")
+	mdb := NewMeshDB(db, db, db, db, lg)
+	layers := NewMesh(mdb, ConfigTst(), &MeshValidatorMock{}, &MockState{}, lg)
 	return layers
 }
 
@@ -47,6 +48,10 @@ func TestLayers_AddBlock(t *testing.T) {
 	block2 := NewBlock(true, []byte("data2"), time.Now(), 2)
 	block3 := NewBlock(true, []byte("data3"), time.Now(), 3)
 
+	addTransactionsToBlock(block1, 4)
+
+	fmt.Println(block1)
+
 	err := layers.AddBlock(block1)
 	assert.NoError(t, err)
 	err = layers.AddBlock(block2)
@@ -57,6 +62,10 @@ func TestLayers_AddBlock(t *testing.T) {
 	rBlock2, err := layers.GetBlock(block2.Id)
 	assert.NoError(t, err)
 
+	rBlock1, err := layers.GetBlock(block1.Id)
+	assert.NoError(t, err)
+
+	assert.True(t, len(rBlock1.Txs) == len(block1.Txs), "block content was wrong")
 	assert.True(t, bytes.Compare(rBlock2.Data, []byte("data2")) == 0, "block content was wrong")
 }