Repository: coinbase/rosetta-cli
Pull request #80 (2020-07-26 18:06:17)

Add BroadcastStorage
Related Issue: #68 

`BroadcastStorage` tracks the broadcast and confirmation of transactions on-chain. It contains the core broadcast logic used in automated construction API testing.

### Changes
- [x] Update `BlockStorage.FindTransaction` function signature based on learnings from `BroadcastStorage`
- [x] Add `BroadcastStorage` implementation
- [x] Add `BroadcastStorage` tests


Patch:
diff --git a/internal/storage/block_storage.go b/internal/storage/block_storage.go
index 64ee203..e11dba6 100644
--- a/internal/storage/block_storage.go
+++ b/internal/storage/block_storage.go
@@ -19,7 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"log"
-	"math"
 
 	"github.com/coinbase/rosetta-sdk-go/syncer"
 	"github.com/coinbase/rosetta-sdk-go/types"
@@ -54,10 +53,6 @@ var (
 	// found in BlockStorage.
 	ErrBlockNotFound = errors.New("block not found")
 
-	// ErrTransactionNotFound is returned when a transaction
-	// is not found in BlockStorage.
-	ErrTransactionNotFound = errors.New("transaction not found")
-
 	// ErrDuplicateBlockHash is returned when a block hash
 	// cannot be stored because it is a duplicate.
 	ErrDuplicateBlockHash = errors.New("duplicate block hash")
@@ -478,57 +473,62 @@ func (b *BlockStorage) removeTransactionHash(
 	return transaction.Set(ctx, hashKey, encodedResult)
 }
 
-// FindTransaction returns the []*types.BlockIdentifier containing the
-// transaction and the depth from the current head of the first transaction
-// sigting (almost always this will just be a single block). If not found,
-// it returns a ErrTransactionNotFound error.
+// FindTransaction returns the most recent *types.BlockIdentifier containing the
+// transaction and the transaction.
 func (b *BlockStorage) FindTransaction(
 	ctx context.Context,
 	transactionIdentifier *types.TransactionIdentifier,
-) ([]*types.BlockIdentifier, int64, error) {
+) (*types.BlockIdentifier, *types.Transaction, error) {
 	txn := b.db.NewDatabaseTransaction(ctx, false)
 	defer txn.Discard(ctx)
 
 	txExists, tx, err := txn.Get(ctx, getTransactionHashKey(transactionIdentifier))
 	if err != nil {
-		return nil, -1, fmt.Errorf("%w: unable to query database for transaction", err)
+		return nil, nil, fmt.Errorf("%w: unable to query database for transaction", err)
 	}
 
 	if !txExists {
-		return nil, -1, nil
+		return nil, nil, nil
 	}
 
 	var blocks map[string]int64
 	if err := decode(tx, &blocks); err != nil {
-		return nil, -1, fmt.Errorf("%w: unable to decode block data for transaction", err)
+		return nil, nil, fmt.Errorf("%w: unable to decode block data for transaction", err)
 	}
 
-	blockExists, block, err := txn.Get(ctx, getHeadBlockKey())
+	var newestBlock *types.BlockIdentifier
+	for hash, index := range blocks {
+		b := &types.BlockIdentifier{Hash: hash, Index: index}
+		if newestBlock == nil || b.Index > newestBlock.Index {
+			newestBlock = b
+		}
+	}
+
+	blockExists, block, err := txn.Get(ctx, getBlockKey(newestBlock))
 	if err != nil {
-		return nil, -1, fmt.Errorf("%w: unable to query database for head block", err)
+		return nil, nil, fmt.Errorf("%w: unable to query database for block", err)
 	}
 
 	if !blockExists {
-		// It should not be possible to enter this conditional
-		// as the existence of a tx hash implies at least
-		// one block is synced.
-		return nil, -1, ErrHeadBlockNotFound
+		return nil, nil, ErrBlockNotFound
 	}
 
-	var head types.BlockIdentifier
-	err = decode(block, &head)
+	var parsedBlock *types.Block
+	err = decode(block, &parsedBlock)
 	if err != nil {
-		return nil, -1, fmt.Errorf("%w: could not decode head block", err)
+		return nil, nil, fmt.Errorf("%w: could not decode block", err)
 	}
 
-	ids := []*types.BlockIdentifier{}
-	oldestBlock := int64(math.MaxInt64)
-	for hash, index := range blocks {
-		ids = append(ids, &types.BlockIdentifier{Hash: hash, Index: index})
-		if index < oldestBlock {
-			oldestBlock = index
+	for _, tx := range parsedBlock.Transactions {
+		if types.Hash(tx.TransactionIdentifier) == types.Hash(transactionIdentifier) {
+			return newestBlock, tx, nil
 		}
 	}
 
-	return ids, head.Index - oldestBlock, nil
+	return nil, nil, fmt.Errorf(
+		"unable to find transaction %s in expected block %s:%d",
+		transactionIdentifier.Hash,
+		newestBlock.Hash,
+		newestBlock.Index,
+	)
 }
diff --git a/internal/storage/broadcast_storage.go b/internal/storage/broadcast_storage.go
new file mode 100644
index 0000000..97aa6c0
--- /dev/null
+++ b/internal/storage/broadcast_storage.go
@@ -0,0 +1,438 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package storage
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+var _ BlockWorker = (*BroadcastStorage)(nil)
+
+const (
+	transactionBroadcastNamespace = "transaction-broadcast"
+)
+
+func getBroadcastKey(transactionIdentifier *types.TransactionIdentifier) []byte {
+	return []byte(
+		fmt.Sprintf("%s/%s", transactionBroadcastNamespace, transactionIdentifier.Hash),
+	)
+}
+
+// BroadcastStorage implements storage methods for managing
+// transaction broadcast.
+type BroadcastStorage struct {
+	db      Database
+	helper  BroadcastStorageHelper
+	handler BroadcastStorageHandler
+
+	confirmationDepth   int64
+	staleDepth          int64
+	broadcastLimit      int
+	broadcastTrailLimit int64
+}
+
+// BroadcastStorageHelper is used by BroadcastStorage to submit transactions
+// and find said transaction in blocks on-chain.
+type BroadcastStorageHelper interface {
+	// CurrentRemoteBlockIdentifier returns the tip reported by a Rosetta
+	// implementation. This is used to determine if we should attempt broadcast.
+	CurrentRemoteBlockIdentifier(
+		context.Context,
+	) (*types.BlockIdentifier, error)
+
+	// CurrentBlockIdentifier is called before transaction broadcast and is used
+	// to determine if a transaction broadcast is stale.
+	CurrentBlockIdentifier(
+		context.Context,
+	) (*types.BlockIdentifier, error) // used to determine if should rebroadcast
+
+	// FindTransaction looks for the provided TransactionIdentifier in processed
+	// blocks and returns the block identifier containing the most recent sighting
+	// and the transaction seen in that block.
+	FindTransaction(
+		context.Context,
+		*types.TransactionIdentifier,
+	) (*types.BlockIdentifier, *types.Transaction, error) // used to confirm
+
+	// BroadcastTransaction broadcasts a transaction to a Rosetta implementation
+	// and returns the *types.TransactionIdentifier returned by the implementation.
+	BroadcastTransaction(
+		context.Context,
+		string,
+	) (*types.TransactionIdentifier, error) // handle initial broadcast + confirm matches provided + rebroadcast if stale
+}
+
+// BroadcastStorageHandler is invoked when a transaction is confirmed on-chain
+// or when a transaction is considered stale.
+type BroadcastStorageHandler interface {
+	// TransactionConfirmed is called when a transaction is observed on-chain for the
+	// last time at a block height < current block height - confirmationDepth.
+	TransactionConfirmed(
+		context.Context,
+		*types.BlockIdentifier,
+		*types.Transaction,
+		[]*types.Operation,
+	) error // can use locked account again + confirm matches intent + update logger
+
+	// TransactionStale is called when a transaction has not yet been
+	// seen on-chain and is considered stale. This occurs when
+	// current block height - last broadcast > staleDepth.
+	TransactionStale(
+		context.Context,
+		*types.TransactionIdentifier,
+	) error // log in counter (rebroadcast should occur here)
+
+	// BroadcastFailed is called when another transaction broadcast would
+	// put it over the provided broadcast limit.
+	BroadcastFailed(
+		context.Context,
+		*types.TransactionIdentifier,
+		[]*types.Operation,
+	) error
+}
+
+// Broadcast is persisted to the db to track transaction broadcast.
+type Broadcast struct {
+	Identifier    *types.TransactionIdentifier `json:"identifier"`
+	Sender        string                       `json:"sender"`
+	Intent        []*types.Operation           `json:"intent"`
+	Payload       string                       `json:"payload"`
+	LastBroadcast *types.BlockIdentifier       `json:"broadcast_at"`
+	Broadcasts    int                          `json:"broadcasts"`
+}
+
+// NewBroadcastStorage returns a new BroadcastStorage.
+func NewBroadcastStorage(
+	db Database,
+	confirmationDepth int64,
+	staleDepth int64,
+	broadcastLimit int,
+	broadcastTrailLimit int64,
+) *BroadcastStorage {
+	return &BroadcastStorage{
+		db:                  db,
+		confirmationDepth:   confirmationDepth,
+		staleDepth:          staleDepth,
+		broadcastLimit:      broadcastLimit,
+		broadcastTrailLimit: broadcastTrailLimit,
+	}
+}
+
+// Initialize adds a BroadcastStorageHelper and BroadcastStorageHandler to BroadcastStorage.
+// This must be called prior to syncing!
+func (b *BroadcastStorage) Initialize(
+	helper BroadcastStorageHelper,
+	handler BroadcastStorageHandler,
+) {
+	b.helper = helper
+	b.handler = handler
+}
+
+func (b *BroadcastStorage) addBlockCommitWorker(
+	ctx context.Context,
+	block *types.Block,
+	staleTransactions []*types.TransactionIdentifier,
+	confirmedTransactions []*Broadcast,
+	foundTransactions []*types.Transaction,
+) error {
+	for _, stale := range staleTransactions {
+		if err := b.handler.TransactionStale(ctx, stale); err != nil {
+			return fmt.Errorf("%w: unable to handle stale transaction %s", err, stale.Hash)
+		}
+	}
+
+	for i, broadcast := range confirmedTransactions {
+		err := b.handler.TransactionConfirmed(
+			ctx,
+			block.BlockIdentifier,
+			foundTransactions[i],
+			broadcast.Intent,
+		)
+		if err != nil {
+			return fmt.Errorf(
+				"%w: unable to handle confirmed transaction %s",
+				err,
+				broadcast.Identifier.Hash,
+			)
+		}
+	}
+
+	if err := b.broadcastPending(ctx); err != nil {
+		return fmt.Errorf("%w: unable to broadcast pending transactions", err)
+	}
+
+	return nil
+}
+
+// AddingBlock is called by BlockStorage when adding a block.
+func (b *BroadcastStorage) AddingBlock(
+	ctx context.Context,
+	block *types.Block,
+	transaction DatabaseTransaction,
+) (CommitWorker, error) {
+	broadcasts, err := b.GetAllBroadcasts(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("%w: unable to get all broadcasts", err)
+	}
+
+	staleTransactions := []*types.TransactionIdentifier{}
+	confirmedTransactions := []*Broadcast{}
+	foundTransactions := []*types.Transaction{}
+
+	for _, broadcast := range broadcasts {
+		if broadcast.LastBroadcast == nil {
+			continue
+		}
+
+		key := getBroadcastKey(broadcast.Identifier)
+
+		foundBlock, foundTransaction, err := b.helper.FindTransaction(ctx, broadcast.Identifier)
+		if err != nil {
+			return nil, fmt.Errorf("%w: unable to determine if transaction was seen", err)
+		}
+
+		// Check if we should mark the broadcast as stale
+		if foundBlock == nil &&
+			block.BlockIdentifier.Index-broadcast.LastBroadcast.Index > b.staleDepth {
+			staleTransactions = append(staleTransactions, broadcast.Identifier)
+			broadcast.LastBroadcast = nil
+			bytes, err := encode(broadcast)
+			if err != nil {
+				return nil, fmt.Errorf("%w: unable to encode updated broadcast", err)
+			}
+
+			if err := transaction.Set(ctx, key, bytes); err != nil {
+				return nil, fmt.Errorf("%w: unable to update broadcast", err)
+			}
+
+			continue
+		}
+
+		// Continue if we are still waiting for a broadcast to appear and it isn't stale
+		if foundBlock == nil {
+			continue
+		}
+
+		// Check if we should mark the transaction as confirmed
+		if block.BlockIdentifier.Index-foundBlock.Index > b.confirmationDepth {
+			confirmedTransactions = append(confirmedTransactions, broadcast)
+			foundTransactions = append(foundTransactions, foundTransaction)
+
+			if err := transaction.Delete(ctx, key); err != nil {
+				return nil, fmt.Errorf("%w: unable to delete confirmed broadcast", err)
+			}
+		}
+	}
+
+	return func(ctx context.Context) error {
+		return b.addBlockCommitWorker(
+			ctx,
+			block,
+			staleTransactions,
+			confirmedTransactions,
+			foundTransactions,
+		)
+	}, nil
+}
+
+// RemovingBlock is called by BlockStorage when removing a block.
+// TODO: error if transaction removed after confirmed (means confirmation depth not deep enough)
+func (b *BroadcastStorage) RemovingBlock(
+	ctx context.Context,
+	block *types.Block,
+	transaction DatabaseTransaction,
+) (CommitWorker, error) {
+	return nil, nil
+}
+
+// Broadcast is called when a caller wants a transaction to be broadcast and tracked.
+// The caller SHOULD NOT broadcast the transaction before calling this function.
+func (b *BroadcastStorage) Broadcast(
+	ctx context.Context,
+	sender string,
+	intent []*types.Operation,
+	transactionIdentifier *types.TransactionIdentifier,
+	payload string,
+) error {
+	txn := b.db.NewDatabaseTransaction(ctx, true)
+	defer txn.Discard(ctx)
+
+	broadcastKey := getBroadcastKey(transactionIdentifier)
+
+	exists, _, err := txn.Get(ctx, broadcastKey)
+	if err != nil {
+		return fmt.Errorf("%w: unable to determine if already broadcasting transaction", err)
+	}
+
+	if exists {
+		return fmt.Errorf("already broadcasting transaction %s", transactionIdentifier.Hash)
+	}
+
+	bytes, err := encode(&Broadcast{
+		Identifier: transactionIdentifier,
+		Sender:     sender,
+		Intent:     intent,
+		Payload:    payload,
+		Broadcasts: 0,
+	})
+	if err != nil {
+		return fmt.Errorf("%w: unable to encode broadcast", err)
+	}
+
+	if err := txn.Set(ctx, broadcastKey, bytes); err != nil {
+		return fmt.Errorf("%w: unable to set broadcast", err)
+	}
+
+	if err := txn.Commit(ctx); err != nil {
+		return fmt.Errorf("%w: unable to commit broadcast", err)
+	}
+
+	return nil
+}
+
+// GetAllBroadcasts returns all currently in-process broadcasts.
+func (b *BroadcastStorage) GetAllBroadcasts(ctx context.Context) ([]*Broadcast, error) {
+	rawBroadcasts, err := b.db.Scan(ctx, []byte(transactionBroadcastNamespace))
+	if err != nil {
+		return nil, fmt.Errorf("%w: unable to scan for all broadcasts", err)
+	}
+
+	broadcasts := make([]*Broadcast, len(rawBroadcasts))
+	for i, rawBroadcast := range rawBroadcasts {
+		var b Broadcast
+		if err := decode(rawBroadcast, &b); err != nil {
+			return nil, fmt.Errorf("%w: unable to decode broadcast", err)
+		}
+
+		broadcasts[i] = &b
+	}
+
+	return broadcasts, nil
+}
+
+func (b *BroadcastStorage) broadcastPending(ctx context.Context) error {
+	tip, err := b.helper.CurrentRemoteBlockIdentifier(ctx)
+	if err != nil {
+		return fmt.Errorf("%w: unable to get current remote block identifier", err)
+	}
+
+	currBlock, err := b.helper.CurrentBlockIdentifier(ctx)
+	if err != nil {
+		return fmt.Errorf("%w: unable to get current block identifier", err)
+	}
+
+	// Wait to broadcast transaction until close to tip
+	if tip.Index-currBlock.Index > b.broadcastTrailLimit {
+		return nil
+	}
+
+	broadcasts, err := b.GetAllBroadcasts(ctx)
+	if err != nil {
+		return fmt.Errorf("%w: unable to get all broadcasts", err)
+	}
+
+	for _, broadcast := range broadcasts {
+		// When a transaction should be broadcast, its last broadcast field must
+		// be set to nil.
+		if broadcast.LastBroadcast != nil {
+			continue
+		}
+
+		if broadcast.Broadcasts >= b.broadcastLimit {
+			txn := b.db.NewDatabaseTransaction(ctx, true)
+			defer txn.Discard(ctx)
+
+			if err := txn.Delete(ctx, getBroadcastKey(broadcast.Identifier)); err != nil {
+				return fmt.Errorf("%w: unable to delete broadcast", err)
+			}
+
+			if err := txn.Commit(ctx); err != nil {
+				return fmt.Errorf("%w: unable to commit broadcast delete", err)
+			}
+
+			if err := b.handler.BroadcastFailed(ctx, broadcast.Identifier, broadcast.Intent); err != nil {
+				return fmt.Errorf("%w: unable to handle broadcast failure", err)
+			}
+
+			continue
+		}
+
+		// We set the last broadcast value before broadcast so we don't accidentally
+		// re-broadcast if exiting between broadcasting the transaction and updating
+		// the value in the database. If the transaction is never really broadcast,
+		// it will be rebroadcast when it is considered stale!
+		broadcast.LastBroadcast = currBlock
+		broadcast.Broadcasts++
+
+		bytes, err := encode(broadcast)
+		if err != nil {
+			return fmt.Errorf("%w: unable to encode broadcast", err)
+		}
+
+		txn := b.db.NewDatabaseTransaction(ctx, true)
+		defer txn.Discard(ctx)
+
+		if err := txn.Set(ctx, getBroadcastKey(broadcast.Identifier), bytes); err != nil {
+			return fmt.Errorf("%w: unable to update broadcast", err)
+		}
+
+		if err := txn.Commit(ctx); err != nil {
+			return fmt.Errorf("%w: unable to commit broadcast update", err)
+		}
+
+		broadcastIdentifier, err := b.helper.BroadcastTransaction(ctx, broadcast.Payload)
+		if err != nil {
+			// Don't error on broadcast failure, retries will automatically be handled.
+			log.Printf(
+				"%s: unable to broadcast transaction %s",
+				err.Error(),
+				broadcast.Identifier.Hash,
+			)
+
+			continue
+		}
+
+		if types.Hash(broadcastIdentifier) != types.Hash(broadcast.Identifier) {
+			return fmt.Errorf(
+				"transaction hash returned by broadcast %s does not match expected %s",
+				broadcastIdentifier.Hash,
+				broadcast.Identifier.Hash,
+			)
+		}
+	}
+
+	return nil
+}
+
+// LockedAddresses returns all addresses currently broadcasting a transaction.
+// The caller SHOULD NOT broadcast a transaction from an account if it is
+// considered locked!
+func (b *BroadcastStorage) LockedAddresses(ctx context.Context) ([]string, error) {
+	broadcasts, err := b.GetAllBroadcasts(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("%w: unable to get all broadcasts", err)
+	}
+
+	addresses := make([]string, len(broadcasts))
+	for i, broadcast := range broadcasts {
+		addresses[i] = broadcast.Sender
+	}
+
+	return addresses, nil
+}


Test patch:
diff --git a/internal/storage/balance_storage_test.go b/internal/storage/balance_storage_test.go
index a479460..4bd7567 100644
--- a/internal/storage/balance_storage_test.go
+++ b/internal/storage/balance_storage_test.go
@@ -607,6 +607,8 @@ func TestBootstrapBalances(t *testing.T) {
 	})
 }
 
+var _ BalanceStorageHelper = (*MockBalanceStorageHelper)(nil)
+
 type MockBalanceStorageHelper struct {
 	AccountBalanceAmount string
 	AccountBalances      map[string]string
diff --git a/internal/storage/block_storage_test.go b/internal/storage/block_storage_test.go
index 49abebc..5d9dd02 100644
--- a/internal/storage/block_storage_test.go
+++ b/internal/storage/block_storage_test.go
@@ -263,13 +263,13 @@ func TestBlock(t *testing.T) {
 	storage := NewBlockStorage(database)
 
 	t.Run("Get non-existent tx", func(t *testing.T) {
-		txBlocks, headDistance, err := storage.FindTransaction(
+		newestBlock, transaction, err := storage.FindTransaction(
 			ctx,
 			newBlock.Transactions[0].TransactionIdentifier,
 		)
 		assert.NoError(t, err)
-		assert.Nil(t, txBlocks)
-		assert.Equal(t, int64(-1), headDistance)
+		assert.Nil(t, newestBlock)
+		assert.Nil(t, transaction)
 	})
 
 	t.Run("Set and get block", func(t *testing.T) {
@@ -284,14 +284,13 @@ func TestBlock(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock.BlockIdentifier, head)
 
-		txBlocks, headDistance, err := storage.FindTransaction(
+		newestBlock, transaction, err := storage.FindTransaction(
 			ctx,
 			newBlock.Transactions[0].TransactionIdentifier,
 		)
 		assert.NoError(t, err)
-		assert.Len(t, txBlocks, 1)
-		assert.Equal(t, newBlock.BlockIdentifier, txBlocks[0])
-		assert.Equal(t, int64(0), headDistance)
+		assert.Equal(t, newBlock.BlockIdentifier, newestBlock)
+		assert.Equal(t, newBlock.Transactions[0], transaction)
 	})
 
 	t.Run("Get non-existent block", func(t *testing.T) {
@@ -321,18 +320,13 @@ func TestBlock(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock2.BlockIdentifier, head)
 
-		txBlocks, headDistance, err := storage.FindTransaction(
+		newestBlock, transaction, err := storage.FindTransaction(
 			ctx,
 			newBlock.Transactions[0].TransactionIdentifier,
 		)
 		assert.NoError(t, err)
-		assert.Len(t, txBlocks, 2)
-		assert.ElementsMatch(
-			t,
-			[]*types.BlockIdentifier{newBlock.BlockIdentifier, newBlock2.BlockIdentifier},
-			txBlocks,
-		)
-		assert.Equal(t, int64(1), headDistance)
+		assert.Equal(t, newBlock2.BlockIdentifier, newestBlock)
+		assert.Equal(t, newBlock2.Transactions[0], transaction)
 	})
 
 	t.Run("Remove block and re-set block of same hash", func(t *testing.T) {
@@ -350,18 +344,13 @@ func TestBlock(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock2.BlockIdentifier, head)
 
-		txBlocks, headDistance, err := storage.FindTransaction(
+		newestBlock, transaction, err := storage.FindTransaction(
 			ctx,
 			newBlock.Transactions[0].TransactionIdentifier,
 		)
 		assert.NoError(t, err)
-		assert.Len(t, txBlocks, 2)
-		assert.ElementsMatch(
-			t,
-			[]*types.BlockIdentifier{newBlock.BlockIdentifier, newBlock2.BlockIdentifier},
-			txBlocks,
-		)
-		assert.Equal(t, int64(1), headDistance)
+		assert.Equal(t, newBlock2.BlockIdentifier, newestBlock)
+		assert.Equal(t, newBlock2.Transactions[0], transaction)
 	})
 
 	t.Run("Add block with complex metadata", func(t *testing.T) {
@@ -375,6 +364,14 @@ func TestBlock(t *testing.T) {
 		head, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, complexBlock.BlockIdentifier, head)
+
+		newestBlock, transaction, err := storage.FindTransaction(
+			ctx,
+			newBlock.Transactions[0].TransactionIdentifier,
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, newBlock2.BlockIdentifier, newestBlock)
+		assert.Equal(t, newBlock2.Transactions[0], transaction)
 	})
 
 	t.Run("Set duplicate transaction hash (same block)", func(t *testing.T) {
diff --git a/internal/storage/broadcast_storage_test.go b/internal/storage/broadcast_storage_test.go
new file mode 100644
index 0000000..c9006c5
--- /dev/null
+++ b/internal/storage/broadcast_storage_test.go
@@ -0,0 +1,860 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package storage
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/coinbase/rosetta-cli/internal/utils"
+
+	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	confirmationDepth   = int64(2)
+	staleDepth          = int64(1)
+	broadcastLimit      = 3
+	broadcastTrailLimit = 10
+)
+
+func blockFiller(start int64, end int64) []*types.Block {
+	blocks := []*types.Block{}
+	for i := start; i < end; i++ {
+		parentIndex := i - 1
+		if parentIndex < 0 {
+			parentIndex = 0
+		}
+		blocks = append(blocks, &types.Block{
+			BlockIdentifier: &types.BlockIdentifier{
+				Index: i,
+				Hash:  fmt.Sprintf("block %d", i),
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Index: parentIndex,
+				Hash:  fmt.Sprintf("block %d", parentIndex),
+			},
+		})
+	}
+
+	return blocks
+}
+
+func opFiller(sender string, opNumber int) []*types.Operation {
+	ops := make([]*types.Operation, opNumber)
+	for i := 0; i < opNumber; i++ {
+		ops[i] = &types.Operation{
+			OperationIdentifier: &types.OperationIdentifier{
+				Index: int64(i),
+			},
+			Account: &types.AccountIdentifier{
+				Address: sender,
+				SubAccount: &types.SubAccountIdentifier{
+					Address: sender,
+				},
+			},
+		}
+	}
+
+	return ops
+}
+
+func TestBroadcastStorageBroadcastSuccess(t *testing.T) {
+	ctx := context.Background()
+
+	newDir, err := utils.CreateTempDir()
+	assert.NoError(t, err)
+	defer utils.RemoveTempDir(newDir)
+
+	database, err := NewBadgerStorage(ctx, newDir)
+	assert.NoError(t, err)
+	defer database.Close(ctx)
+
+	storage := NewBroadcastStorage(
+		database,
+		confirmationDepth,
+		staleDepth,
+		broadcastLimit,
+		broadcastTrailLimit,
+	)
+	mockHelper := &MockBroadcastStorageHelper{}
+	mockHandler := &MockBroadcastStorageHandler{}
+	storage.Initialize(mockHelper, mockHandler)
+
+	send1 := opFiller("addr 1", 11)
+	send2 := opFiller("addr 2", 13)
+	mockHelper.Transactions = map[string]*types.TransactionIdentifier{
+		"payload 1": {Hash: "tx 1"},
+		"payload 2": {Hash: "tx 2"},
+	}
+
+	t.Run("broadcast send 1", func(t *testing.T) {
+		err := storage.Broadcast(
+			ctx,
+			"addr 1",
+			send1,
+			&types.TransactionIdentifier{Hash: "tx 1"},
+			"payload 1",
+		)
+		assert.NoError(t, err)
+
+		// Check to make sure duplicate instances of address aren't reported
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 1)
+		assert.ElementsMatch(t, []string{"addr 1"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:     "addr 1",
+				Intent:     send1,
+				Payload:    "payload 1",
+			},
+		}, broadcasts)
+	})
+
+	blocks := blockFiller(0, 7)
+	t.Run("add block 0", func(t *testing.T) {
+		block := blocks[0]
+		mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+		txn := storage.db.NewDatabaseTransaction(ctx, true)
+		commitWorker, err := storage.AddingBlock(ctx, block, txn)
+		assert.NoError(t, err)
+		err = txn.Commit(ctx)
+		assert.NoError(t, err)
+
+		mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+		err = commitWorker(ctx)
+		assert.NoError(t, err)
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 1)
+		assert.ElementsMatch(t, []string{"addr 1"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:        "addr 1",
+				Intent:        send1,
+				Payload:       "payload 1",
+				LastBroadcast: blocks[0].BlockIdentifier,
+				Broadcasts:    1,
+			},
+		}, broadcasts)
+
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+	})
+
+	t.Run("broadcast send 2", func(t *testing.T) {
+		err := storage.Broadcast(
+			ctx,
+			"addr 2",
+			send2,
+			&types.TransactionIdentifier{Hash: "tx 2"},
+			"payload 2",
+		)
+		assert.NoError(t, err)
+
+		// Check to make sure duplicate instances of address aren't reported
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:        "addr 1",
+				Intent:        send1,
+				Payload:       "payload 1",
+				LastBroadcast: blocks[0].BlockIdentifier,
+				Broadcasts:    1,
+			},
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:     "addr 2",
+				Intent:     send2,
+				Payload:    "payload 2",
+			},
+		}, broadcasts)
+	})
+
+	t.Run("add block 1", func(t *testing.T) {
+		block := blocks[1]
+		mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+		txn := storage.db.NewDatabaseTransaction(ctx, true)
+		commitWorker, err := storage.AddingBlock(ctx, block, txn)
+		assert.NoError(t, err)
+		err = txn.Commit(ctx)
+		assert.NoError(t, err)
+
+		mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+		err = commitWorker(ctx)
+		assert.NoError(t, err)
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:        "addr 1",
+				Intent:        send1,
+				Payload:       "payload 1",
+				LastBroadcast: blocks[0].BlockIdentifier,
+				Broadcasts:    1,
+			},
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:        "addr 2",
+				Intent:        send2,
+				Payload:       "payload 2",
+				LastBroadcast: blocks[1].BlockIdentifier,
+				Broadcasts:    1,
+			},
+		}, broadcasts)
+
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+	})
+
+	t.Run("add block 2", func(t *testing.T) {
+		block := blocks[2]
+		mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+		txn := storage.db.NewDatabaseTransaction(ctx, true)
+		commitWorker, err := storage.AddingBlock(ctx, block, txn)
+		assert.NoError(t, err)
+		err = txn.Commit(ctx)
+		assert.NoError(t, err)
+
+		mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+		err = commitWorker(ctx)
+		assert.NoError(t, err)
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:        "addr 1",
+				Intent:        send1,
+				Payload:       "payload 1",
+				LastBroadcast: blocks[2].BlockIdentifier,
+				Broadcasts:    2,
+			},
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:        "addr 2",
+				Intent:        send2,
+				Payload:       "payload 2",
+				LastBroadcast: blocks[1].BlockIdentifier,
+				Broadcasts:    1,
+			},
+		}, broadcasts)
+
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{
+			{
+				Hash: "tx 1",
+			},
+		}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+	})
+
+	tx1 := &types.Transaction{
+		TransactionIdentifier: &types.TransactionIdentifier{Hash: "tx 1"},
+		Operations:            send1,
+	}
+	tx2 := &types.Transaction{
+		TransactionIdentifier: &types.TransactionIdentifier{Hash: "tx 2"},
+		Operations:            send2,
+	}
+	t.Run("add block 3", func(t *testing.T) {
+		block := blocks[3]
+		block.Transactions = []*types.Transaction{tx1, tx2}
+		mockHelper.FindTransactions = map[string]*findTx{
+			tx1.TransactionIdentifier.Hash: {
+				blockIdentifier: block.BlockIdentifier,
+				transaction:     tx1,
+			},
+			tx2.TransactionIdentifier.Hash: {
+				blockIdentifier: block.BlockIdentifier,
+				transaction:     tx2,
+			},
+		}
+		mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+		txn := storage.db.NewDatabaseTransaction(ctx, true)
+		commitWorker, err := storage.AddingBlock(ctx, block, txn)
+		assert.NoError(t, err)
+		err = txn.Commit(ctx)
+		assert.NoError(t, err)
+
+		mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+		err = commitWorker(ctx)
+		assert.NoError(t, err)
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:        "addr 1",
+				Intent:        send1,
+				Payload:       "payload 1",
+				LastBroadcast: blocks[2].BlockIdentifier,
+				Broadcasts:    2,
+			},
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:        "addr 2",
+				Intent:        send2,
+				Payload:       "payload 2",
+				LastBroadcast: blocks[1].BlockIdentifier,
+				Broadcasts:    1,
+			},
+		}, broadcasts)
+
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{
+			{
+				Hash: "tx 1",
+			},
+		}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+		assert.ElementsMatch(t, []*confirmedTx{}, mockHandler.Confirmed)
+	})
+
+	t.Run("add block 4-5", func(t *testing.T) {
+		for _, block := range blocks[4:6] {
+			mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+			txn := storage.db.NewDatabaseTransaction(ctx, true)
+			commitWorker, err := storage.AddingBlock(ctx, block, txn)
+			assert.NoError(t, err)
+			err = txn.Commit(ctx)
+			assert.NoError(t, err)
+
+			mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+			err = commitWorker(ctx)
+			assert.NoError(t, err)
+
+			addresses, err := storage.LockedAddresses(ctx)
+			assert.NoError(t, err)
+			assert.Len(t, addresses, 2)
+			assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+			broadcasts, err := storage.GetAllBroadcasts(ctx)
+			assert.NoError(t, err)
+			assert.ElementsMatch(t, []*Broadcast{
+				{
+					Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+					Sender:        "addr 1",
+					Intent:        send1,
+					Payload:       "payload 1",
+					LastBroadcast: blocks[2].BlockIdentifier,
+					Broadcasts:    2,
+				},
+				{
+					Identifier:    &types.TransactionIdentifier{Hash: "tx 2"},
+					Sender:        "addr 2",
+					Intent:        send2,
+					Payload:       "payload 2",
+					LastBroadcast: blocks[1].BlockIdentifier,
+					Broadcasts:    1,
+				},
+			}, broadcasts)
+			assert.ElementsMatch(t, []*types.TransactionIdentifier{
+				{
+					Hash: "tx 1",
+				},
+			}, mockHandler.Stale)
+			assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+			assert.ElementsMatch(t, []*confirmedTx{}, mockHandler.Confirmed)
+		}
+	})
+
+	t.Run("add block 6", func(t *testing.T) {
+		block := blocks[6]
+		mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+		txn := storage.db.NewDatabaseTransaction(ctx, true)
+		commitWorker, err := storage.AddingBlock(ctx, block, txn)
+		assert.NoError(t, err)
+		err = txn.Commit(ctx)
+		assert.NoError(t, err)
+
+		mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+		err = commitWorker(ctx)
+		assert.NoError(t, err)
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 0)
+		assert.ElementsMatch(t, []string{}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{}, broadcasts)
+
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{
+			{
+				Hash: "tx 1",
+			},
+		}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+		assert.ElementsMatch(t, []*confirmedTx{
+			{
+				blockIdentifier: block.BlockIdentifier,
+				transaction:     tx1,
+				intent:          send1,
+			},
+			{
+				blockIdentifier: block.BlockIdentifier,
+				transaction:     tx2,
+				intent:          send2,
+			},
+		}, mockHandler.Confirmed)
+	})
+}
+
+func TestBroadcastStorageBroadcastFailure(t *testing.T) {
+	ctx := context.Background()
+
+	newDir, err := utils.CreateTempDir()
+	assert.NoError(t, err)
+	defer utils.RemoveTempDir(newDir)
+
+	database, err := NewBadgerStorage(ctx, newDir)
+	assert.NoError(t, err)
+	defer database.Close(ctx)
+
+	storage := NewBroadcastStorage(
+		database,
+		confirmationDepth,
+		staleDepth,
+		broadcastLimit,
+		broadcastTrailLimit,
+	)
+	mockHelper := &MockBroadcastStorageHelper{}
+	mockHandler := &MockBroadcastStorageHandler{}
+	storage.Initialize(mockHelper, mockHandler)
+
+	t.Run("locked addresses with no broadcasts", func(t *testing.T) {
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 0)
+		assert.ElementsMatch(t, []string{}, addresses)
+	})
+
+	send1 := opFiller("addr 1", 11)
+	send2 := opFiller("addr 2", 13)
+	t.Run("broadcast", func(t *testing.T) {
+		err := storage.Broadcast(
+			ctx,
+			"addr 1",
+			send1,
+			&types.TransactionIdentifier{Hash: "tx 1"},
+			"payload 1",
+		)
+		assert.NoError(t, err)
+
+		err = storage.Broadcast(
+			ctx,
+			"addr 2",
+			send2,
+			&types.TransactionIdentifier{Hash: "tx 2"},
+			"payload 2",
+		)
+		assert.NoError(t, err)
+
+		// Check to make sure duplicate instances of address aren't reported
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:     "addr 1",
+				Intent:     send1,
+				Payload:    "payload 1",
+			},
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:     "addr 2",
+				Intent:     send2,
+				Payload:    "payload 2",
+			},
+		}, broadcasts)
+	})
+
+	t.Run("add blocks and expire", func(t *testing.T) {
+		mockHelper.Transactions = map[string]*types.TransactionIdentifier{
+			"payload 1": {Hash: "tx 1"},
+			// payload 2 will fail
+		}
+		blocks := blockFiller(0, 10)
+		for _, block := range blocks {
+			mockHelper.RemoteBlockIdentifier = block.BlockIdentifier
+
+			txn := storage.db.NewDatabaseTransaction(ctx, true)
+			commitWorker, err := storage.AddingBlock(ctx, block, txn)
+			assert.NoError(t, err)
+			err = txn.Commit(ctx)
+			assert.NoError(t, err)
+
+			mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+			err = commitWorker(ctx)
+			assert.NoError(t, err)
+		}
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 0)
+		assert.ElementsMatch(t, []string{}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{}, broadcasts)
+
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{
+			{Hash: "tx 1"},
+			{Hash: "tx 1"},
+			{Hash: "tx 1"},
+			{Hash: "tx 2"},
+			{Hash: "tx 2"},
+			{Hash: "tx 2"},
+		}, mockHandler.Stale)
+
+		assert.ElementsMatch(t, []*failedTx{
+			{
+				transaction: &types.TransactionIdentifier{Hash: "tx 1"},
+				intent:      send1,
+			},
+			{
+				transaction: &types.TransactionIdentifier{Hash: "tx 2"},
+				intent:      send2,
+			},
+		}, mockHandler.Failed)
+
+		assert.ElementsMatch(t, []*confirmedTx{}, mockHandler.Confirmed)
+	})
+}
+
+func TestBroadcastStorageBehindTip(t *testing.T) {
+	ctx := context.Background()
+
+	newDir, err := utils.CreateTempDir()
+	assert.NoError(t, err)
+	defer utils.RemoveTempDir(newDir)
+
+	database, err := NewBadgerStorage(ctx, newDir)
+	assert.NoError(t, err)
+	defer database.Close(ctx)
+
+	storage := NewBroadcastStorage(
+		database,
+		confirmationDepth,
+		staleDepth,
+		broadcastLimit,
+		broadcastTrailLimit,
+	)
+	mockHelper := &MockBroadcastStorageHelper{}
+	mockHandler := &MockBroadcastStorageHandler{}
+	storage.Initialize(mockHelper, mockHandler)
+
+	send1 := opFiller("addr 1", 11)
+	send2 := opFiller("addr 2", 13)
+	t.Run("broadcast", func(t *testing.T) {
+		err := storage.Broadcast(
+			ctx,
+			"addr 1",
+			send1,
+			&types.TransactionIdentifier{Hash: "tx 1"},
+			"payload 1",
+		)
+		assert.NoError(t, err)
+
+		err = storage.Broadcast(
+			ctx,
+			"addr 2",
+			send2,
+			&types.TransactionIdentifier{Hash: "tx 2"},
+			"payload 2",
+		)
+		assert.NoError(t, err)
+
+		// Check to make sure duplicate instances of address aren't reported
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:     "addr 1",
+				Intent:     send1,
+				Payload:    "payload 1",
+			},
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:     "addr 2",
+				Intent:     send2,
+				Payload:    "payload 2",
+			},
+		}, broadcasts)
+	})
+
+	blocks := blockFiller(0, 81)
+	mockHelper.RemoteBlockIdentifier = blocks[80].BlockIdentifier
+	mockHelper.Transactions = map[string]*types.TransactionIdentifier{
+		"payload 1": {Hash: "tx 1"},
+		"payload 2": {Hash: "tx 2"},
+	}
+
+	t.Run("add blocks behind tip", func(t *testing.T) {
+		for _, block := range blocks[:60] {
+			txn := storage.db.NewDatabaseTransaction(ctx, true)
+			commitWorker, err := storage.AddingBlock(ctx, block, txn)
+			assert.NoError(t, err)
+			err = txn.Commit(ctx)
+			assert.NoError(t, err)
+
+			mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+			err = commitWorker(ctx)
+			assert.NoError(t, err)
+		}
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:     "addr 1",
+				Intent:     send1,
+				Payload:    "payload 1",
+			},
+			{
+				Identifier: &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:     "addr 2",
+				Intent:     send2,
+				Payload:    "payload 2",
+			},
+		}, broadcasts)
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+		assert.ElementsMatch(t, []*confirmedTx{}, mockHandler.Confirmed)
+	})
+
+	t.Run("add blocks close to tip", func(t *testing.T) {
+		for _, block := range blocks[60:71] {
+			txn := storage.db.NewDatabaseTransaction(ctx, true)
+			commitWorker, err := storage.AddingBlock(ctx, block, txn)
+			assert.NoError(t, err)
+			err = txn.Commit(ctx)
+			assert.NoError(t, err)
+
+			mockHelper.SyncedBlockIdentifier = block.BlockIdentifier
+			err = commitWorker(ctx)
+			assert.NoError(t, err)
+		}
+
+		addresses, err := storage.LockedAddresses(ctx)
+		assert.NoError(t, err)
+		assert.Len(t, addresses, 2)
+		assert.ElementsMatch(t, []string{"addr 1", "addr 2"}, addresses)
+
+		broadcasts, err := storage.GetAllBroadcasts(ctx)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []*Broadcast{
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 1"},
+				Sender:        "addr 1",
+				Intent:        send1,
+				Payload:       "payload 1",
+				LastBroadcast: blocks[70].BlockIdentifier,
+				Broadcasts:    1,
+			},
+			{
+				Identifier:    &types.TransactionIdentifier{Hash: "tx 2"},
+				Sender:        "addr 2",
+				Intent:        send2,
+				Payload:       "payload 2",
+				LastBroadcast: blocks[70].BlockIdentifier,
+				Broadcasts:    1,
+			},
+		}, broadcasts)
+		assert.ElementsMatch(t, []*types.TransactionIdentifier{}, mockHandler.Stale)
+		assert.ElementsMatch(t, []*failedTx{}, mockHandler.Failed)
+		assert.ElementsMatch(t, []*confirmedTx{}, mockHandler.Confirmed)
+	})
+}
+
+var _ BroadcastStorageHelper = (*MockBroadcastStorageHelper)(nil)
+
+type findTx struct {
+	blockIdentifier *types.BlockIdentifier
+	transaction     *types.Transaction
+}
+
+type MockBroadcastStorageHelper struct {
+	RemoteBlockIdentifier *types.BlockIdentifier
+	SyncedBlockIdentifier *types.BlockIdentifier
+	Transactions          map[string]*types.TransactionIdentifier
+	FindTransactions      map[string]*findTx
+}
+
+func (m *MockBroadcastStorageHelper) CurrentRemoteBlockIdentifier(
+	ctx context.Context,
+) (*types.BlockIdentifier, error) {
+	return m.RemoteBlockIdentifier, nil
+}
+
+func (m *MockBroadcastStorageHelper) CurrentBlockIdentifier(
+	ctx context.Context,
+) (*types.BlockIdentifier, error) {
+	return m.SyncedBlockIdentifier, nil
+}
+
+func (m *MockBroadcastStorageHelper) FindTransaction(
+	ctx context.Context,
+	transactionIdentifier *types.TransactionIdentifier,
+) (*types.BlockIdentifier, *types.Transaction, error) {
+	val, exists := m.FindTransactions[transactionIdentifier.Hash]
+	if !exists {
+		return nil, nil, nil
+	}
+
+	return val.blockIdentifier, val.transaction, nil
+}
+
+func (m *MockBroadcastStorageHelper) BroadcastTransaction(
+	ctx context.Context,
+	payload string,
+) (*types.TransactionIdentifier, error) {
+	val, exists := m.Transactions[payload]
+	if !exists {
+		return nil, errors.New("broadcast error")
+	}
+
+	return val, nil
+}
+
+var _ BroadcastStorageHandler = (*MockBroadcastStorageHandler)(nil)
+
+type confirmedTx struct {
+	blockIdentifier *types.BlockIdentifier
+	transaction     *types.Transaction
+	intent          []*types.Operation
+}
+
+type failedTx struct {
+	transaction *types.TransactionIdentifier
+	intent      []*types.Operation
+}
+
+type MockBroadcastStorageHandler struct {
+	Confirmed []*confirmedTx
+	Stale     []*types.TransactionIdentifier
+	Failed    []*failedTx
+}
+
+func (m *MockBroadcastStorageHandler) TransactionConfirmed(
+	ctx context.Context,
+	blockIdentifier *types.BlockIdentifier,
+	transaction *types.Transaction,
+	intent []*types.Operation,
+) error {
+	if m.Confirmed == nil {
+		m.Confirmed = []*confirmedTx{}
+	}
+
+	m.Confirmed = append(m.Confirmed, &confirmedTx{
+		blockIdentifier: blockIdentifier,
+		transaction:     transaction,
+		intent:          intent,
+	})
+
+	return nil
+}
+
+func (m *MockBroadcastStorageHandler) TransactionStale(
+	ctx context.Context,
+	transactionIdentifier *types.TransactionIdentifier,
+) error {
+	if m.Stale == nil {
+		m.Stale = []*types.TransactionIdentifier{}
+	}
+
+	m.Stale = append(m.Stale, transactionIdentifier)
+
+	return nil
+}
+
+func (m *MockBroadcastStorageHandler) BroadcastFailed(
+	ctx context.Context,
+	transactionIdentifier *types.TransactionIdentifier,
+	intent []*types.Operation,
+) error {
+	if m.Failed == nil {
+		m.Failed = []*failedTx{}
+	}
+
+	m.Failed = append(m.Failed, &failedTx{
+		transaction: transactionIdentifier,
+		intent:      intent,
+	})
+
+	return nil
+}