Only unpublish a dataset on the provider.

dsp/statemachine/contract_statemachine_test.go

@@ -338,12 +338,6 @@ func TestStateMachineConsumerInitConsumerPull(t *testing.T) {
 		},
 	}, nil)
 
-	mockProvider.On("UnpublishDataset", mock.Anything, &providerv1.UnpublishDatasetRequest{
-		PublishId: trProviderPID.String(),
-	}).Return(&providerv1.UnpublishDatasetResponse{
-		Success: true,
-	}, nil)
-
 	apply, err = pTransNext.Send(ctx)
 	assert.Nil(t, err)
 	apply()
@@ -395,6 +389,19 @@ func TestStateMachineConsumerInitConsumerPull(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, trCPID.URN(), trCompletionMSG.ConsumerPID)
 	assert.Equal(t, trProviderPID.URN(), trCompletionMSG.ProviderPID)
+
+	mockProvider.On("UnpublishDataset", mock.Anything, &providerv1.UnpublishDatasetRequest{
+		PublishId: trProviderPID.String(),
+	}).Return(&providerv1.UnpublishDatasetResponse{
+		Success: true,
+	}, nil)
+
+	pTransContractStarted, err := store.GetProviderTransfer(ctx, trProviderPID)
+	validateTransfer(t, err, pTransContractStarted, statemachine.TransferRequestStates.STARTED, agreementID)
+
+	pTransStarted := statemachine.GetTransferRequestNegotiation(store, pTransContractStarted, mockProvider, reconciler)
+	pTransNext, err = pTransStarted.Recv(ctx, trCompletionMSG)
+	validateTransfer(t, err, pTransNext.GetTransferRequest(), statemachine.TransferRequestStates.COMPLETED, agreementID)
 }
 
 // TestContractStateMachineConsumerInit tests a whole contract statemachine run, this will do the happy path,

dsp/statemachine/transfer_request_transitions.go

@@ -147,33 +147,46 @@ func (tr *TransferRequestNegotiationStarted) Recv(
 ) (TransferRequestNegotiationState, error) {
 	switch t := message.(type) {
 	case shared.TransferCompletionMessage:
+		err := unpublishTransfer(ctx, tr)
+		if err != nil {
+			return nil, err
+		}
 		return verifyAndTransformTransfer(ctx, tr, t.ProviderPID, t.ConsumerPID, TransferRequestStates.COMPLETED)
 	case shared.TransferTerminationMessage:
 		return verifyAndTransformTransfer(ctx, tr, t.ProviderPID, t.ConsumerPID, TransferRequestStates.TRANSFERTERMINATED)
-
 	default:
 		return nil, fmt.Errorf("invalid message type")
 	}
 }
 
 func (tr *TransferRequestNegotiationStarted) Send(ctx context.Context) (func(), error) {
+	err := unpublishTransfer(ctx, tr)
+	if err != nil {
+		return func() {}, err
+	}
+	return sendTransferCompletion(ctx, tr)
+}
+
+func unpublishTransfer(ctx context.Context, tr TransferRequestNegotiationState) error {
 	switch tr.GetTransferDirection() {
 	case DirectionPull:
-		_, err := tr.GetProvider().UnpublishDataset(ctx, &providerv1.UnpublishDatasetRequest{
-			PublishId: tr.GetProviderPID().String(),
-		})
-		if err != nil {
-			return func() {}, err
+		if tr.GetRole() == DataspaceProvider {
+			_, err := tr.GetProvider().UnpublishDataset(ctx, &providerv1.UnpublishDatasetRequest{
+				PublishId: tr.GetProviderPID().String(),
+			})
+			if err != nil {
+				return err
+			}
 		}
 	case DirectionPush:
-		// TODO: Signal provider to start uploading dataset here.
-		return func() {}, fmt.Errorf("push flow: %w", ErrNotImplemented)
+
+		return fmt.Errorf("push flow: %w", ErrNotImplemented)
 	case DirectionUnknown:
-		return func() {}, fmt.Errorf("unknown transfer direction")
+		return fmt.Errorf("unknown transfer direction")
 	default:
 		panic("unexpected statemachine.TransferDirection")
 	}
-	return sendTransferCompletion(ctx, tr)
+	return nil
 }
 
 type TransferRequestNegotiationSuspended struct {