Correct data transfer Sent stats (#106)

* correct data transfer stats

* feat(channels): add queued event

Co-authored-by: hannahhoward <[email protected]>

Author: aarshkshah1992

channels/channel_state.go

@@ -34,6 +34,8 @@ type channelState struct {
 	status datatransfer.Status
 	// isPull indicates if this is a push or pull request
 	isPull bool
+	// total bytes read from this node and queued for sending (0 if receiver)
+	queued uint64
 	// total bytes sent from this node (0 if receiver)
 	sent uint64
 	// total bytes received by this node (0 if sender)
@@ -56,6 +58,9 @@ var EmptyChannelState = channelState{}
 // Status is the current status of this channel
 func (c channelState) Status() datatransfer.Status { return c.status }
 
+// Received returns the number of bytes received
+func (c channelState) Queued() uint64 { return c.queued }
+
 // Sent returns the number of bytes sent
 func (c channelState) Sent() uint64 { return c.sent }
 
@@ -171,6 +176,7 @@ func fromInternalChannelState(c internal.ChannelState, voucherDecoder DecoderByT
 		recipient:            c.Recipient,
 		totalSize:            c.TotalSize,
 		status:               c.Status,
+		queued:               c.Queued,
 		sent:                 c.Sent,
 		received:             c.Received,
 		message:              c.Message,

channels/channels.go

@@ -190,6 +190,10 @@ func (c *Channels) DataSent(chid datatransfer.ChannelID, cid cid.Cid, delta uint
 	return c.send(chid, datatransfer.DataSent, delta, cid)
 }
 
+func (c *Channels) DataQueued(chid datatransfer.ChannelID, cid cid.Cid, delta uint64) error {
+	return c.send(chid, datatransfer.DataQueued, delta, cid)
+}
+
 func (c *Channels) DataReceived(chid datatransfer.ChannelID, cid cid.Cid, delta uint64) error {
 	return c.send(chid, datatransfer.DataReceived, delta, cid)
 }

channels/channels_fsm.go

@@ -48,7 +48,17 @@ var ChannelEvents = fsm.Events{
 		chst.Sent += delta
 		return nil
 	}),
-
+	fsm.Event(datatransfer.DataQueued).FromMany(
+		datatransfer.Requested,
+		datatransfer.Ongoing,
+		datatransfer.InitiatorPaused,
+		datatransfer.ResponderPaused,
+		datatransfer.BothPaused,
+		datatransfer.ResponderCompleted,
+		datatransfer.ResponderFinalizing).ToNoChange().Action(func(chst *internal.ChannelState, delta uint64, c cid.Cid) error {
+		chst.Queued += delta
+		return nil
+	}),
 	fsm.Event(datatransfer.Disconnected).FromAny().ToNoChange().Action(func(chst *internal.ChannelState) error {
 		chst.Message = datatransfer.ErrDisconnected.Error()
 		return nil

channels/internal/internalchannel.go

@@ -48,6 +48,8 @@ type ChannelState struct {
 	TotalSize uint64
 	// current status of this deal
 	Status datatransfer.Status
+	// total bytes read from this node and queued for sending (0 if receiver)
+	Queued uint64
 	// total bytes sent from this node (0 if receiver)
 	Sent uint64
 	// total bytes received by this node (0 if sender)

channels/internal/internalchannel_cbor_gen.go

@@ -70,6 +70,9 @@ const (
 	// CompleteCleanupOnRestart is emitted when a data transfer channel is restarted to signal
 	// that channels that were cleaning up should finish cleanup
 	CompleteCleanupOnRestart
+
+	// DataQueued is emmited is read and queued for sending to the remote peer
+	DataQueued
 )
 
 // Events are human readable names for data transfer events
@@ -93,6 +96,7 @@ var Events = map[EventCode]string{
 	BeginFinalizing:             "BeginFinalizing",
 	Complete:                    "Complete",
 	CompleteCleanupOnRestart:    "CompleteCleanupOnRestart",
+	DataQueued:                  "DataQueued",
 }
 
 // Event is a struct containing information about a data transfer event

events.go

@@ -58,9 +58,8 @@ func (m *manager) OnDataReceived(chid datatransfer.ChannelID, link ipld.Link, si
 	return nil
 }
 
-func (m *manager) OnDataSent(chid datatransfer.ChannelID, link ipld.Link, size uint64) (datatransfer.Message, error) {
-	err := m.channels.DataSent(chid, link.(cidlink.Link).Cid, size)
-	if err != nil {
+func (m *manager) OnDataQueued(chid datatransfer.ChannelID, link ipld.Link, size uint64) (datatransfer.Message, error) {
+	if err := m.channels.DataQueued(chid, link.(cidlink.Link).Cid, size); err != nil {
 		return nil, err
 	}
 	if chid.Initiator != m.peerID {
@@ -79,9 +78,14 @@ func (m *manager) OnDataSent(chid datatransfer.ChannelID, link ipld.Link, size u
 			return m.processRevalidationResult(chid, result, err)
 		}
 	}
+
 	return nil, nil
 }
 
+func (m *manager) OnDataSent(chid datatransfer.ChannelID, link ipld.Link, size uint64) error {
+	return m.channels.DataSent(chid, link.(cidlink.Link).Cid, size)
+}
+
 func (m *manager) OnRequestReceived(chid datatransfer.ChannelID, request datatransfer.Request) (datatransfer.Response, error) {
 	if request.IsRestart() {
 		return m.receiveRestartRequest(chid, request)

impl/events.go

@@ -117,9 +117,9 @@ func TestRoundTrip(t *testing.T) {
 				sent := make(chan uint64, 21)
 				received := make(chan uint64, 21)
 				var subscriber datatransfer.Subscriber = func(event datatransfer.Event, channelState datatransfer.ChannelState) {
-					if event.Code == datatransfer.DataSent {
-						if channelState.Sent() > 0 {
-							sent <- channelState.Sent()
+					if event.Code == datatransfer.DataQueued {
+						if channelState.Queued() > 0 {
+							sent <- channelState.Queued()
 						}
 					}
 
@@ -776,7 +776,7 @@ func TestPauseAndResume(t *testing.T) {
 	}
 	for testCase, isPull := range testCases {
 		t.Run(testCase, func(t *testing.T) {
-			ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
 			defer cancel()
 
 			gsData := testutil.NewGraphsyncTestingData(ctx, t, nil, nil)
@@ -805,9 +805,9 @@ func TestPauseAndResume(t *testing.T) {
 			resumeResponder := make(chan struct{}, 2)
 			var subscriber datatransfer.Subscriber = func(event datatransfer.Event, channelState datatransfer.ChannelState) {
 
-				if event.Code == datatransfer.DataSent {
-					if channelState.Sent() > 0 {
-						sent <- channelState.Sent()
+				if event.Code == datatransfer.DataQueued {
+					if channelState.Queued() > 0 {
+						sent <- channelState.Queued()
 					}
 				}
 

impl/integration_test.go

@@ -395,7 +395,7 @@ func TestDataTransferResponding(t *testing.T) {
 				datatransfer.Open,
 				datatransfer.NewVoucherResult,
 				datatransfer.Accept,
-				datatransfer.DataSent,
+				datatransfer.DataQueued,
 				datatransfer.NewVoucherResult,
 				datatransfer.PauseResponder,
 				datatransfer.NewVoucher,
@@ -415,7 +415,7 @@ func TestDataTransferResponding(t *testing.T) {
 			verify: func(t *testing.T, h *receiverHarness) {
 				_, err := h.transport.EventHandler.OnRequestReceived(channelID(h.id, h.peers), h.pullRequest)
 				require.NoError(t, err)
-				msg, err := h.transport.EventHandler.OnDataSent(
+				msg, err := h.transport.EventHandler.OnDataQueued(
 					channelID(h.id, h.peers),
 					cidlink.Link{Cid: testutil.GenerateCids(1)[0]},
 					12345)

impl/responding_test.go

@@ -101,9 +101,9 @@ func TestRestartPush(t *testing.T) {
 			disConnChan := make(chan struct{})
 
 			var subscriber datatransfer.Subscriber = func(event datatransfer.Event, channelState datatransfer.ChannelState) {
-				if event.Code == datatransfer.DataSent {
-					if channelState.Sent() > 0 {
-						sent <- channelState.Sent()
+				if event.Code == datatransfer.DataQueued {
+					if channelState.Queued() > 0 {
+						sent <- channelState.Queued()
 					}
 				}
 
@@ -289,9 +289,9 @@ func TestRestartPull(t *testing.T) {
 			disConnChan := make(chan struct{})
 
 			var subscriber datatransfer.Subscriber = func(event datatransfer.Event, channelState datatransfer.ChannelState) {
-				if event.Code == datatransfer.DataSent {
-					if channelState.Sent() > 0 {
-						sent <- channelState.Sent()
+				if event.Code == datatransfer.DataQueued {
+					if channelState.Queued() > 0 {
+						sent <- channelState.Queued()
 					}
 				}
 

impl/restart_integration_test.go

@@ -26,14 +26,19 @@ type EventsHandler interface {
 	// - error = cancel this request
 	// - err == ErrPause - pause this request
 	OnDataReceived(chid ChannelID, link ipld.Link, size uint64) error
-	// OnDataSent is called when we send data for the given channel ID
+
+	// OnDataQueued is called when data is queued for sending for the given channel ID
 	// return values are:
 	// message = data transfer message along with data
 	// err = error
 	// - nil = proceed with sending data
 	// - error = cancel this request
 	// - err == ErrPause - pause this request
-	OnDataSent(chid ChannelID, link ipld.Link, size uint64) (Message, error)
+	OnDataQueued(chid ChannelID, link ipld.Link, size uint64) (Message, error)
+
+	// OnDataSent is called when we send data for the given channel ID
+	OnDataSent(chid ChannelID, link ipld.Link, size uint64) error
+
 	// OnRequestReceived is called when we receive a new request to send data
 	// for the given channel ID
 	// return values are:

transport.go

@@ -297,6 +297,9 @@ func (t *Transport) SetEventHandler(events datatransfer.EventsHandler) error {
 	t.gs.RegisterCompletedResponseListener(t.gsCompletedResponseListener)
 	t.gs.RegisterIncomingBlockHook(t.gsIncomingBlockHook)
 	t.gs.RegisterOutgoingBlockHook(t.gsOutgoingBlockHook)
+
+	t.gs.RegisterBlockSentListener(t.gsBlockSentHook)
+
 	t.gs.RegisterOutgoingRequestHook(t.gsOutgoingRequestHook)
 	t.gs.RegisterIncomingResponseHook(t.gsIncomingResponseHook)
 	t.gs.RegisterRequestUpdatedHook(t.gsRequestUpdatedHook)
@@ -374,6 +377,19 @@ func (t *Transport) gsIncomingBlockHook(p peer.ID, response graphsync.ResponseDa
 	}
 }
 
+func (t *Transport) gsBlockSentHook(p peer.ID, request graphsync.RequestData, block graphsync.BlockData) {
+	t.dataLock.RLock()
+	chid, ok := t.graphsyncRequestMap[graphsyncKey{request.ID(), p}]
+	t.dataLock.RUnlock()
+	if !ok {
+		return
+	}
+
+	if err := t.events.OnDataSent(chid, block.Link(), block.BlockSize()); err != nil {
+		log.Errorf("failed to process data sent: %+v", err)
+	}
+}
+
 func (t *Transport) gsOutgoingBlockHook(p peer.ID, request graphsync.RequestData, block graphsync.BlockData, hookActions graphsync.OutgoingBlockHookActions) {
 	t.dataLock.RLock()
 	chid, ok := t.graphsyncRequestMap[graphsyncKey{request.ID(), p}]
@@ -389,7 +405,7 @@ func (t *Transport) gsOutgoingBlockHook(p peer.ID, request graphsync.RequestData
 	}
 	rp.maximumSent = rp.currentSent
 
-	msg, err := t.events.OnDataSent(chid, block.Link(), block.BlockSize())
+	msg, err := t.events.OnDataQueued(chid, block.Link(), block.BlockSize())
 	if err != nil && err != datatransfer.ErrPause {
 		hookActions.TerminateWithError(err)
 		return