Fix/close event

channels/channel_state.go

@@ -144,7 +144,7 @@ func (c channelState) InitiatorPaused() bool {
 }
 
 func (c channelState) ResponderPaused() bool {
-	return c.ic.ResponderPaused || c.ic.Status == datatransfer.Finalizing
+	return c.ic.ResponderPaused
 }
 
 func (c channelState) BothPaused() bool {
@@ -158,6 +158,28 @@ func (c channelState) SelfPaused() bool {
 	return c.ResponderPaused()
 }
 
+func (c channelState) TransferClosed() bool {
+	return c.ic.TransferClosed
+}
+
+func (c channelState) ExceededDataLimit() bool {
+	var limitFactor uint64
+	if c.ic.SelfPeer == c.ic.Sender {
+		limitFactor = c.ic.Queued
+	} else {
+		limitFactor = c.ic.Received
+	}
+	return c.ic.DataLimit != 0 && limitFactor >= c.ic.DataLimit
+}
+
+func (c channelState) AwaitingFinalization() bool {
+	return c.Status().InFinalization() && c.ic.RequiresFinalization
+}
+
+func (c channelState) TransferOnHold() bool {
+	return c.SelfPaused() || c.AwaitingFinalization() || c.ExceededDataLimit()
+}
+
 // Stages returns the current ChannelStages object, or an empty object.
 // It is unsafe for the caller to modify the return value, and changes may not
 // be persisted. It should be treated as immutable.

channels/channels.go

@@ -306,6 +306,11 @@ func (c *Channels) SetRequiresFinalization(chid datatransfer.ChannelID, Requires
 	return c.send(chid, datatransfer.SetRequiresFinalization, RequiresFinalization)
 }
 
+// CloseTransfer indicates the transfer is closed, even if the transfer was not finished
+func (c *Channels) CloseTransfer(chid datatransfer.ChannelID) error {
+	return c.send(chid, datatransfer.CloseTransfer)
+}
+
 // HasChannel returns true if the given channel id is being tracked
 func (c *Channels) HasChannel(chid datatransfer.ChannelID) (bool, error) {
 	return c.stateMachines.Has(chid)

channels/channels_fsm.go

@@ -46,6 +46,7 @@ var ChannelEvents = fsm.Events{
 	}),
 
 	fsm.Event(datatransfer.Cancel).FromAny().To(datatransfer.Cancelling).Action(func(chst *internal.ChannelState) error {
+		chst.TransferClosed = true
 		chst.AddLog("")
 		return nil
 	}),
@@ -145,6 +146,7 @@ var ChannelEvents = fsm.Events{
 
 	fsm.Event(datatransfer.Error).FromAny().To(datatransfer.Failing).Action(func(chst *internal.ChannelState, err error) error {
 		chst.Message = err.Error()
+		chst.TransferClosed = true
 		chst.AddLog("data transfer erred: %s", chst.Message)
 		return nil
 	}),
@@ -173,7 +175,7 @@ var ChannelEvents = fsm.Events{
 	// seems less than ideal. We need some kind of support for pausing being an independent aspect of state
 	// Possibly we should just remove whether a state is paused from the state entirely.
 	fsm.Event(datatransfer.PauseInitiator).
-		FromMany(datatransfer.Ongoing, datatransfer.Requested, datatransfer.Queued, datatransfer.AwaitingAcceptance).ToJustRecord().
+		FromMany(datatransfer.Ongoing, datatransfer.Requested, datatransfer.Queued, datatransfer.AwaitingAcceptance, datatransfer.ResponderFinalizing, datatransfer.ResponderFinalizingTransferFinished).ToJustRecord().
 		Action(func(chst *internal.ChannelState) error {
 			chst.InitiatorPaused = true
 			chst.AddLog("")
@@ -191,7 +193,6 @@ var ChannelEvents = fsm.Events{
 	fsm.Event(datatransfer.DataLimitExceeded).
 		FromMany(datatransfer.Ongoing, datatransfer.Requested, datatransfer.Queued, datatransfer.AwaitingAcceptance, datatransfer.ResponderCompleted, datatransfer.ResponderFinalizing).ToJustRecord().
 		Action(func(chst *internal.ChannelState) error {
-			chst.ResponderPaused = true
 			chst.AddLog("")
 			return nil
 		}),
@@ -224,6 +225,7 @@ var ChannelEvents = fsm.Events{
 		// the finalization process and complete the transfer
 		From(datatransfer.AwaitingAcceptance).To(datatransfer.Completing).
 		Action(func(chst *internal.ChannelState) error {
+			chst.TransferClosed = true
 			chst.AddLog("")
 			return nil
 		}),
@@ -248,12 +250,20 @@ var ChannelEvents = fsm.Events{
 	}),
 
 	fsm.Event(datatransfer.BeginFinalizing).FromAny().To(datatransfer.Finalizing).Action(func(chst *internal.ChannelState) error {
+		chst.TransferClosed = true
+		chst.AddLog("")
+		return nil
+	}),
+
+	fsm.Event(datatransfer.CloseTransfer).FromAny().ToJustRecord().Action(func(chst *internal.ChannelState) error {
+		chst.TransferClosed = true
 		chst.AddLog("")
 		return nil
 	}),
 
 	// Both the local node and the remote peer have completed the transfer
 	fsm.Event(datatransfer.Complete).FromAny().To(datatransfer.Completing).Action(func(chst *internal.ChannelState) error {
+		chst.TransferClosed = true
 		chst.AddLog("")
 		return nil
 	}),

channels/channels_test.go

@@ -196,7 +196,6 @@ func TestChannels(t *testing.T) {
 		err = channelList.DataLimitExceeded(datatransfer.ChannelID{Initiator: peers[1], Responder: peers[0], ID: tid1})
 		require.NoError(t, err)
 		state = checkEvent(ctx, t, received, datatransfer.DataLimitExceeded)
-		require.True(t, state.ResponderPaused())
 
 		err = channelList.SetDataLimit(datatransfer.ChannelID{Initiator: peers[1], Responder: peers[0], ID: tid1}, 700)
 		require.NoError(t, err)
@@ -205,7 +204,6 @@ func TestChannels(t *testing.T) {
 
 		err = channelList.ResumeResponder(datatransfer.ChannelID{Initiator: peers[1], Responder: peers[0], ID: tid1})
 		state = checkEvent(ctx, t, received, datatransfer.ResumeResponder)
-		require.False(t, state.ResponderPaused())
 
 		err = channelList.PauseInitiator(datatransfer.ChannelID{Initiator: peers[1], Responder: peers[0], ID: tid1})
 		state = checkEvent(ctx, t, received, datatransfer.PauseInitiator)
@@ -214,7 +212,7 @@ func TestChannels(t *testing.T) {
 		err = channelList.DataLimitExceeded(datatransfer.ChannelID{Initiator: peers[1], Responder: peers[0], ID: tid1})
 		require.NoError(t, err)
 		state = checkEvent(ctx, t, received, datatransfer.DataLimitExceeded)
-		require.True(t, state.BothPaused())
+		require.True(t, state.InitiatorPaused())
 
 	})
 
@@ -440,6 +438,15 @@ func TestMigrations(t *testing.T) {
 		datatransfer.ResponderPaused,
 		datatransfer.BothPaused,
 		datatransfer.Ongoing,
+		datatransfer.TransferFinished,
+		datatransfer.ResponderFinalizingTransferFinished,
+		datatransfer.Finalizing,
+		datatransfer.Completed,
+		datatransfer.Completing,
+		datatransfer.Failing,
+		datatransfer.Failed,
+		datatransfer.Cancelling,
+		datatransfer.Cancelled,
 	}
 	for i := 0; i < numChannels; i++ {
 		transferIDs[i] = datatransfer.TransferID(rand.Uint64())
@@ -515,8 +522,9 @@ func TestMigrations(t *testing.T) {
 		datatransfer.Ongoing,
 	}
 
-	expectedInitiatorPaused := []bool{false, true, false, true, false}
-	expectedResponderPaused := []bool{false, false, true, true, false}
+	expectedInitiatorPaused := []bool{false, true, false, true, false, false, false, false, false, false, false, false, false, false}
+	expectedResponderPaused := []bool{false, false, true, true, false, false, false, false, false, false, false, false, false, false}
+	expectedTransferClosed := []bool{false, false, false, false, false, true, true, true, true, true, true, true, true, true}
 	for i := 0; i < numChannels; i++ {
 
 		channel, err := channelList.GetByID(ctx, datatransfer.ChannelID{
@@ -540,10 +548,10 @@ func TestMigrations(t *testing.T) {
 		require.Equal(t, expectedStatuses[i], channel.Status())
 		require.Equal(t, expectedInitiatorPaused[i], channel.InitiatorPaused())
 		require.Equal(t, expectedResponderPaused[i], channel.ResponderPaused())
+		require.Equal(t, expectedTransferClosed[i], channel.TransferClosed())
 		require.Equal(t, basicnode.NewInt(sentIndex[i]), channel.SentIndex())
 		require.Equal(t, basicnode.NewInt(receivedIndex[i]), channel.ReceivedIndex())
 		require.Equal(t, basicnode.NewInt(queuedIndex[i]), channel.QueuedIndex())
-
 	}
 }
 

channels/internal/internalchannel.go

@@ -121,6 +121,8 @@ type ChannelState struct {
 	ResponderPaused bool
 	// InitiatorPaused indicates whether the initiator is in a paused state
 	InitiatorPaused bool
+	// TransferClosed indicates the transfer portion of the request is over
+	TransferClosed bool
 	// Stages traces the execution fo a data transfer.
 	//
 	// EXPERIMENTAL; subject to change.

channels/internal/migrations/migrations.go

@@ -71,6 +71,18 @@ func NoOpChannelState0To2(oldChannelState *ChannelStateV2) (*ChannelStateV2, err
 	return oldChannelState, nil
 }
 
+var transferClosedStatuses = []datatransfer.Status{
+	datatransfer.TransferFinished,
+	datatransfer.ResponderFinalizingTransferFinished,
+	datatransfer.Finalizing,
+	datatransfer.Completed,
+	datatransfer.Completing,
+	datatransfer.Failing,
+	datatransfer.Failed,
+	datatransfer.Cancelling,
+	datatransfer.Cancelled,
+}
+
 func MigrateChannelState2To3(oldChannelState *ChannelStateV2) (*internal.ChannelState, error) {
 	receivedIndex := basicnode.NewInt(oldChannelState.ReceivedBlocksTotal)
 	sentIndex := basicnode.NewInt(oldChannelState.SentBlocksTotal)
@@ -82,6 +94,12 @@ func MigrateChannelState2To3(oldChannelState *ChannelStateV2) (*internal.Channel
 	if newStatus == datatransfer.ResponderPaused || newStatus == datatransfer.InitiatorPaused || newStatus == datatransfer.BothPaused {
 		newStatus = datatransfer.Ongoing
 	}
+	var transferClosed bool
+	for _, status := range transferClosedStatuses {
+		if oldChannelState.Status == status {
+			transferClosed = true
+		}
+	}
 	return &internal.ChannelState{
 		SelfPeer:             oldChannelState.SelfPeer,
 		TransferID:           oldChannelState.TransferID,
@@ -106,6 +124,7 @@ func MigrateChannelState2To3(oldChannelState *ChannelStateV2) (*internal.Channel
 		RequiresFinalization: oldChannelState.RequiresFinalization,
 		InitiatorPaused:      initiatorPaused,
 		ResponderPaused:      responderPaused,
+		TransferClosed:       transferClosed,
 		Stages:               oldChannelState.Stages,
 	}, nil
 }

events.go

@@ -129,6 +129,9 @@ const (
 
 	// SendMessageError indicates an error sending a data transfer message
 	SendMessageError
+
+	// CloseTransfer is called indicates the transfer is complete, but not because it was finished
+	CloseTransfer
 )
 
 // Events are human readable names for data transfer events
@@ -169,6 +172,7 @@ var Events = map[EventCode]string{
 	DataLimitExceeded:           "DataLimitExceeded",
 	TransferInitiated:           "TransferInitiated",
 	SendMessageError:            "SendMessageError",
+	CloseTransfer:               "CloseTransfer",
 }
 
 // Event is a struct containing information about a data transfer event

impl/events.go

@@ -42,7 +42,11 @@ func (m *manager) processTransferEvent(ctx context.Context, chid datatransfer.Ch
 			return err
 		}
 		msg := message.UpdateResponse(chid.ID, true)
-		return m.transport.SendMessage(ctx, chid, msg)
+		err := m.transport.SendMessage(ctx, chid, msg)
+		if err != nil {
+			return m.channels.SendMessageError(chid, err)
+		}
+		return nil
 	case datatransfer.TransportTransferCancelled:
 		log.Warnf("channel %+v was cancelled: %s", chid, evt.ErrorMessage)
 		return m.channels.RequestCancelled(chid, errors.New(evt.ErrorMessage))
@@ -168,7 +172,7 @@ func (m *manager) channelCompleted(chid datatransfer.ChannelID, success bool, er
 	// If the transferred errored on completion
 	if !success {
 		// send an error, but only if we haven't already errored/finished transfer already for some reason
-		if !chst.Status().TransferComplete() {
+		if !chst.TransferClosed() {
 			err := fmt.Errorf("data transfer channel %s failed to transfer data: %s", chid, errorMessage)
 			log.Warnf(err.Error())
 			return m.channels.Error(chid, err)