Store the count of unique Receieved cids

channelmonitor/channelmonitor_test.go

@@ -544,6 +544,10 @@ func (m *mockChannelState) Status() datatransfer.Status {
 	return datatransfer.Ongoing
 }
 
+func (m *mockChannelState) NReceivedCids() uint64 {
+	panic("implement me")
+}
+
 func (m *mockChannelState) TransferID() datatransfer.TransferID {
 	panic("implement me")
 }

channels/channel_state.go

@@ -50,6 +50,8 @@ type channelState struct {
 	voucherDecoder       DecoderByTypeFunc
 	channelCIDsReader    ChannelCIDsReader
 
+	nReceivedCids uint64
+
 	// stages tracks the timeline of events related to a data transfer, for
 	// traceability purposes.
 	stages *datatransfer.ChannelStages
@@ -107,6 +109,10 @@ func (c channelState) ReceivedCids() []cid.Cid {
 	return receivedCids
 }
 
+func (c channelState) NReceivedCids() uint64 {
+	return c.nReceivedCids
+}
+
 // Sender returns the peer id for the node that is sending data
 func (c channelState) Sender() peer.ID { return c.sender }
 
@@ -211,6 +217,7 @@ func fromInternalChannelState(c internal.ChannelState, voucherDecoder DecoderByT
 		voucherDecoder:       voucherDecoder,
 		channelCIDsReader:    channelCIDsReader,
 		stages:               c.Stages,
+		nReceivedCids:        c.NReceivedCids,
 	}
 }
 

channels/channels_fsm.go

@@ -50,6 +50,7 @@ var ChannelEvents = fsm.Events{
 	fsm.Event(datatransfer.DataReceivedProgress).FromMany(transferringStates...).ToNoChange().
 		Action(func(chst *internal.ChannelState, delta uint64) error {
 			chst.Received += delta
+			chst.NReceivedCids = chst.NReceivedCids + 1
 			chst.AddLog("received data")
 			return nil
 		}),

channels/channels_test.go

@@ -143,6 +143,7 @@ func TestChannels(t *testing.T) {
 		require.Equal(t, datatransfer.Requested, state.Status())
 		require.Equal(t, uint64(0), state.Received())
 		require.Equal(t, uint64(0), state.Sent())
+
 		require.Empty(t, state.ReceivedCids())
 
 		isNew, err := channelList.DataReceived(datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1}, cids[0], 50)
@@ -153,6 +154,7 @@ func TestChannels(t *testing.T) {
 		require.Equal(t, uint64(50), state.Received())
 		require.Equal(t, uint64(0), state.Sent())
 		require.Equal(t, []cid.Cid{cids[0]}, state.ReceivedCids())
+		require.Equal(t, uint64(1), state.NReceivedCids())
 
 		isNew, err = channelList.DataSent(datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1}, cids[1], 100)
 		require.NoError(t, err)
@@ -180,6 +182,7 @@ func TestChannels(t *testing.T) {
 		require.Equal(t, uint64(100), state.Received())
 		require.Equal(t, uint64(100), state.Sent())
 		require.Equal(t, []cid.Cid{cids[0], cids[1]}, state.ReceivedCids())
+		require.Equal(t, uint64(2), state.NReceivedCids())
 
 		isNew, err = channelList.DataSent(datatransfer.ChannelID{Initiator: peers[0], Responder: peers[1], ID: tid1}, cids[1], 25)
 		require.NoError(t, err)

channels/internal/internalchannel.go

@@ -61,6 +61,9 @@ type ChannelState struct {
 	Vouchers       []EncodedVoucher
 	VoucherResults []EncodedVoucherResult
 
+	// NReceivedCids is the number of cids for which data has already been received by this node.
+	NReceivedCids uint64
+
 	// Stages traces the execution fo a data transfer.
 	//
 	// EXPERIMENTAL; subject to change.

types.go

@@ -132,6 +132,9 @@ type ChannelState interface {
 	// ReceivedCids returns the cids received so far on the channel
 	ReceivedCids() []cid.Cid
 
+	// NReceivedCids returns the number of unique cids received on the channel.
+	NReceivedCids() uint64
+
 	// Queued returns the number of bytes read from the node and queued for sending
 	Queued() uint64