fix(Stripe): Routinely save state in Stripe payments workflow (#584)

* fix(Stripe): Routinely save state in Stripe payments workflow

* Increase workflow timeout to 5 minutes

* Convert StartToCloseTimeoutMinutes value to constant

* Only update the activity timeout for FetchNext.* activities

* Define activity timeout consts in models to avoid import cycle

Author: laouji

internal/connectors/engine/workflow/context.go

@@ -3,13 +3,26 @@ package workflow
 import (
 	"time"
 
+	"github.com/formancehq/payments/internal/models"
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 )
 
 func infiniteRetryContext(ctx workflow.Context) workflow.Context {
 	return workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-		StartToCloseTimeout: 60 * time.Second,
+		StartToCloseTimeout: models.ActivityStartToCloseTimeoutMinutesDefault * time.Minute,
+		RetryPolicy: &temporal.RetryPolicy{
+			InitialInterval:        time.Second,
+			BackoffCoefficient:     2,
+			MaximumInterval:        100 * time.Second,
+			NonRetryableErrorTypes: []string{},
+		},
+	})
+}
+
+func fetchNextActivityRetryContext(ctx workflow.Context) workflow.Context {
+	return workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+		StartToCloseTimeout: models.ActivityStartToCloseTimeoutMinutesLong * time.Minute,
 		RetryPolicy: &temporal.RetryPolicy{
 			InitialInterval:        time.Second,
 			BackoffCoefficient:     2,

internal/connectors/engine/workflow/fetch_accounts.go

@@ -52,7 +52,7 @@ func (w Workflow) fetchAccounts(
 	hasMore := true
 	for hasMore {
 		accountsResponse, err := activities.PluginFetchNextAccounts(
-			infiniteRetryContext(ctx),
+			fetchNextActivityRetryContext(ctx),
 			fetchNextAccount.ConnectorID,
 			fetchNextAccount.FromPayload.GetPayload(),
 			state.State,

internal/connectors/engine/workflow/fetch_balances.go

@@ -52,7 +52,7 @@ func (w Workflow) fetchBalances(
 	hasMore := true
 	for hasMore {
 		balancesResponse, err := activities.PluginFetchNextBalances(
-			infiniteRetryContext(ctx),
+			fetchNextActivityRetryContext(ctx),
 			fetchNextBalances.ConnectorID,
 			fetchNextBalances.FromPayload.GetPayload(),
 			state.State,

internal/connectors/engine/workflow/fetch_external_accounts.go

@@ -52,7 +52,7 @@ func (w Workflow) fetchExternalAccounts(
 	hasMore := true
 	for hasMore {
 		externalAccountsResponse, err := activities.PluginFetchNextExternalAccounts(
-			infiniteRetryContext(ctx),
+			fetchNextActivityRetryContext(ctx),
 			fetchNextExternalAccount.ConnectorID,
 			fetchNextExternalAccount.FromPayload.GetPayload(),
 			state.State,

internal/connectors/engine/workflow/fetch_others.go

@@ -51,7 +51,7 @@ func (w Workflow) fetchNextOthers(
 	hasMore := true
 	for hasMore {
 		othersResponse, err := activities.PluginFetchNextOthers(
-			infiniteRetryContext(ctx),
+			fetchNextActivityRetryContext(ctx),
 			fetchNextOthers.ConnectorID,
 			fetchNextOthers.Name,
 			fetchNextOthers.FromPayload.GetPayload(),

internal/connectors/engine/workflow/fetch_payments.go

@@ -52,7 +52,7 @@ func (w Workflow) fetchNextPayments(
 	hasMore := true
 	for hasMore {
 		paymentsResponse, err := activities.PluginFetchNextPayments(
-			infiniteRetryContext(ctx),
+			fetchNextActivityRetryContext(ctx),
 			fetchNextPayments.ConnectorID,
 			fetchNextPayments.FromPayload.GetPayload(),
 			state.State,

internal/connectors/plugins/public/stripe/client/payments.go

@@ -3,8 +3,10 @@ package client
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/formancehq/payments/internal/connectors/metrics"
+	"github.com/formancehq/payments/internal/models"
 	"github.com/stripe/stripe-go/v79"
 )
 
@@ -27,11 +29,17 @@ func (c *client) GetPayments(
 ) (results []*stripe.BalanceTransaction, _ Timeline, hasMore bool, err error) {
 	results = make([]*stripe.BalanceTransaction, 0, int(pageSize))
 
+	timer := time.NewTimer((models.ActivityStartToCloseTimeoutMinutesLong - 1) * time.Minute)
+	defer timer.Stop()
+
 	for {
 		select {
 		case <-ctx.Done():
 			// if the app is shutting down
 			return results, timeline, false, fmt.Errorf("context closed before first payment found")
+		case <-timer.C:
+			// after the timer expires let's save the state to prevent workflow timeout
+			return results, timeline, true, nil
 		default: //fallthrough
 		}
 

internal/models/activity_context.go

@@ -0,0 +1,6 @@
+package models
+
+const (
+	ActivityStartToCloseTimeoutMinutesDefault = 1
+	ActivityStartToCloseTimeoutMinutesLong    = 5
+)