disable HandleTransactionAnnouncement

cmd/arc/services/metamorph.go

@@ -3,9 +3,6 @@ package cmd
 import (
 	"context"
 	"fmt"
-	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_core"
-	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_jetstream"
-	"github.com/bitcoin-sv/arc/internal/message_queue/nats/nats_connection"
 	"log/slog"
 	"net"
 	"net/http"
@@ -14,6 +11,10 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_core"
+	"github.com/bitcoin-sv/arc/internal/message_queue/nats/client/nats_jetstream"
+	"github.com/bitcoin-sv/arc/internal/message_queue/nats/nats_connection"
+
 	"github.com/bitcoin-sv/arc/config"
 	"github.com/bitcoin-sv/arc/internal/blocktx/blocktx_api"
 	"github.com/bitcoin-sv/arc/internal/metamorph"

internal/metamorph/peer_handler.go

@@ -45,15 +45,6 @@ func (m *PeerHandler) HandleTransactionSent(msg *wire.MsgTx, peer p2p.PeerI) err
 
 // HandleTransactionAnnouncement is a message sent to the PeerHandler when a transaction INV message is received from a peer.
 func (m *PeerHandler) HandleTransactionAnnouncement(msg *wire.InvVect, peer p2p.PeerI) error {
-	select {
-	case m.messageCh <- &PeerTxMessage{
-		Hash:   &msg.Hash,
-		Status: metamorph_api.Status_SEEN_ON_NETWORK,
-		Peer:   peer.String(),
-	}:
-	default: // Ensure that writing to channel is non-blocking
-	}
-
 	return nil
 }
 

internal/metamorph/peer_handler_test.go

@@ -53,31 +53,6 @@ func TestPeerHandler(t *testing.T) {
 		}
 	})
 
-	t.Run("HandleTransactionAnnouncement", func(t *testing.T) {
-		hash, err := chainhash.NewHashFromStr("1234")
-		require.NoError(t, err)
-
-		msgInv := wire.NewInvVect(wire.InvTypeBlock, hash)
-		require.NoError(t, err)
-
-		expectedMsg := &metamorph.PeerTxMessage{
-			Hash:   &msgInv.Hash,
-			Status: metamorph_api.Status_SEEN_ON_NETWORK,
-			Peer:   "mock_peer",
-		}
-
-		go func() {
-			_ = peerHandler.HandleTransactionAnnouncement(msgInv, peer)
-		}()
-
-		select {
-		case msg := <-messageCh:
-			assert.Equal(t, expectedMsg, msg)
-		case <-time.After(time.Second):
-			t.Fatal("test timed out or error while executing goroutine")
-		}
-	})
-
 	t.Run("HandleTransactionRejection", func(t *testing.T) {
 		msgReject := wire.NewMsgReject("command", wire.RejectMalformed, "malformed")
 

internal/metamorph/processor.go

@@ -86,6 +86,14 @@ type Processor struct {
 
 	processMinedInterval  time.Duration
 	processMinedBatchSize int
+
+	orderedDnnouncedTransactions []AnnouncedTransaction
+	announcedTransactionsLock    sync.Mutex
+}
+
+type AnnouncedTransaction struct {
+	second uint64
+	hash   *chainhash.Hash
 }
 
 type Option func(f *Processor)
@@ -120,8 +128,9 @@ func NewProcessor(s store.MetamorphStore, pm p2p.PeerManagerI, statusMessageChan
 		maxRetries:                maxRetriesDefault,
 		minimumHealthyConnections: minimumHealthyConnectionsDefault,
 
-		responseProcessor: NewResponseProcessor(),
-		statusMessageCh:   statusMessageChannel,
+		responseProcessor:            NewResponseProcessor(),
+		statusMessageCh:              statusMessageChannel,
+		orderedDnnouncedTransactions: make([]AnnouncedTransaction, 0),
 
 		processExpiredTxsInterval:       unseenTransactionRebroadcastingInterval,
 		processSeenOnNetworkTxsInterval: seenOnNetworkTransactionRequestingInterval,
@@ -197,6 +206,7 @@ func (p *Processor) Start() error {
 	p.StartProcessExpiredTransactions()
 	p.StartRequestingSeenOnNetworkTxs()
 	p.StartProcessStatusUpdatesInStorage()
+	p.StartCheckingTransactionsInNetwork()
 	p.StartProcessMinedCallbacks()
 	err = p.StartCollectStats()
 	if err != nil {
@@ -384,6 +394,38 @@ func (p *Processor) StartSendStatusUpdate() {
 	}()
 }
 
+func (p *Processor) StartCheckingTransactionsInNetwork() {
+	p.waitGroup.Add(1)
+	ticker := time.NewTicker(100 * time.Millisecond)
+	go func() {
+		defer p.waitGroup.Done()
+		for {
+			select {
+			case <-p.ctx.Done():
+				return
+
+			case <-ticker.C:
+				p.announcedTransactionsLock.Lock()
+				for k := 0; k < len(p.orderedDnnouncedTransactions); k++ {
+					if p.orderedDnnouncedTransactions[k].second < uint64(p.now().Unix())-1 {
+						p.logger.Info("requested transaction", slog.String("hash", p.orderedDnnouncedTransactions[k].hash.String()))
+						p.pm.RequestTransaction((p.orderedDnnouncedTransactions[k].hash))
+						if k == len(p.orderedDnnouncedTransactions)-1 {
+							p.orderedDnnouncedTransactions = []AnnouncedTransaction{}
+							break
+						}
+					} else {
+						p.orderedDnnouncedTransactions = p.orderedDnnouncedTransactions[k:]
+						break
+					}
+				}
+				p.announcedTransactionsLock.Unlock()
+				ticker.Reset(100 * time.Millisecond)
+			}
+		}
+	}()
+}
+
 func (p *Processor) StartProcessStatusUpdatesInStorage() {
 	ticker := time.NewTicker(p.processStatusUpdatesInterval)
 	p.waitGroup.Add(1)
@@ -678,9 +720,6 @@ func (p *Processor) ProcessTransaction(ctx context.Context, req *ProcessorReques
 		Status: metamorph_api.Status_STORED,
 	})
 
-	// Send GETDATA to peers to see if they have it
-	p.pm.RequestTransaction(req.Data.Hash)
-
 	// Announce transaction to network peers
 	p.logger.Debug("announcing transaction", slog.String("hash", req.Data.Hash.String()))
 	peers := p.pm.AnnounceTransaction(req.Data.Hash, nil)
@@ -689,6 +728,9 @@ func (p *Processor) ProcessTransaction(ctx context.Context, req *ProcessorReques
 		return
 	}
 
+	// will be requesting transaction after ~2 seconds to get SEEN_ON_NETWORK status
+	p.DeferRequestTransaction(req.Data.Hash)
+
 	// update status in response
 	statusResponse.UpdateStatus(StatusAndError{
 		Status: metamorph_api.Status_ANNOUNCED_TO_NETWORK,
@@ -704,6 +746,15 @@ func (p *Processor) ProcessTransaction(ctx context.Context, req *ProcessorReques
 	p.responseProcessor.Add(statusResponse)
 }
 
+func (p *Processor) DeferRequestTransaction(txHash *chainhash.Hash) {
+	p.announcedTransactionsLock.Lock()
+	p.orderedDnnouncedTransactions = append(p.orderedDnnouncedTransactions, AnnouncedTransaction{
+		second: uint64(p.now().Unix()),
+		hash:   txHash,
+	})
+	p.announcedTransactionsLock.Unlock()
+}
+
 func (p *Processor) ProcessTransactions(sReq []*store.StoreData) {
 	// store in database
 	err := p.store.SetBulk(p.ctx, sReq)

internal/metamorph/processor_test.go

@@ -153,7 +153,7 @@ func TestProcessTransaction(t *testing.T) {
 			expectedResponseMapItems: 0,
 			expectedSetCalls:         1,
 			expectedAnnounceCalls:    1,
-			expectedRequestCalls:     1,
+			expectedRequestCalls:     0,
 		},
 		{
 			name: "record found",
@@ -943,6 +943,50 @@ func TestProcessorHealth(t *testing.T) {
 	}
 }
 
+func TestStartCheckingTransactionsInNetwork(t *testing.T) {
+	metamorphStore := &storeMocks.MetamorphStoreMock{
+		GetFunc: func(ctx context.Context, key []byte) (*store.StoreData, error) {
+			return &store.StoreData{Hash: testdata.TX2Hash}, nil
+		},
+		SetUnlockedByNameFunc: func(ctx context.Context, lockedBy string) (int64, error) { return 0, nil },
+		GetUnminedFunc: func(ctx context.Context, since time.Time, limit int64, offset int64) ([]*store.StoreData, error) {
+			return nil, nil
+		},
+		IncrementRetriesFunc: func(ctx context.Context, hash *chainhash.Hash) error {
+			return nil
+		},
+	}
+	pm := &mocks.PeerManagerMock{
+		RequestTransactionFunc: func(txHash *chainhash.Hash) p2p.PeerI {
+			return nil
+		},
+		ShutdownFunc: func() {},
+	}
+
+	publisher := &mocks.MessageQueueClientMock{
+		PublishFunc: func(topic string, hash []byte) error {
+			return nil
+		},
+	}
+
+	processor, err := metamorph.NewProcessor(metamorphStore, pm, nil, metamorph.WithMessageQueueClient(publisher), metamorph.WithProcessExpiredTxsInterval(time.Millisecond*20), metamorph.WithMaxRetries(10), metamorph.WithNow(func() time.Time {
+		return time.Date(2033, 1, 1, 1, 0, 0, 0, time.UTC)
+	}))
+	require.NoError(t, err)
+	defer processor.Shutdown()
+
+	processor.StartCheckingTransactionsInNetwork()
+
+	require.Equal(t, 0, processor.GetProcessorMapSize())
+
+	processor.DeferRequestTransaction(testdata.TX2Hash)
+
+	time.Sleep(1 * time.Second)
+	require.Equal(t, 0, len(pm.RequestTransactionCalls()))
+	time.Sleep(2 * time.Second)
+	require.Equal(t, 1, len(pm.RequestTransactionCalls()))
+}
+
 func TestStart(t *testing.T) {
 	tt := []struct {
 		name     string