mirror of
https://github.com/hanzo-fi/payments.git
synced 2026-10-30 20:31:20 +00:00
5d9e69dbc1
* fix(workflow): EN-1093 generate reset connector reference via SideEffect
resetConnector built the new connector with Reference: uuid.New() directly in
the workflow body. Temporal replays workflow code, and the reset awaits a
potentially long uninstall child workflow, so on replay a different UUID was
produced: the install child WorkflowID install-{stack}-{newConnector.ID} no
longer matched history, the workflow task failed the determinism check in a
loop and the reset stayed stuck.
Generate the reference through workflow.SideEffect, which records the value in
history and returns the same UUID on replay.
Constraint: Temporal replays workflow code; the body must be deterministic
Rejected: Generate the reference engine-side and pass it as workflow input | larger blast radius (ResetConnector struct + engine call site + all reset tests) for no extra safety
Confidence: high
Scope-risk: narrow
Not-tested: replay determinism itself (no WorkflowReplayer/recorded-history harness in repo); existing TestWorkflowEnvironment executes once without replay
* fix(workflow): EN-1093 read connector polling period via a deterministic activity
scheduleNextWorkflow / create_payout / create_transfer read the polling period from
the in-process connectors.Manager (w.connectors.GetConfig) inside the workflow body and
fed it into the TemporalScheduleCreate activity's arguments, which are recorded in
history. A polling-period change via UpdateConnector (which does not terminate running
workflows) could therefore make an already-scheduled workflow non-deterministic on
replay — recomputing a different interval than history — leaving it permanently stuck.
The value also depended on which pod's in-memory config ran the task.
Read the polling period through a new StorageConnectorsGetPollingPeriod activity (DB,
recorded in history), gated by workflow.GetVersion so in-flight workflows keep the
legacy in-process read until they drain. The activity parses onto configurer
DefaultConfig (so an absent pollingPeriod resolves to the default, not a zero-interval
schedule) and returns only the duration.
Constraint: prod Temporal payload encryption is OFF — the activity returns only the polling period, never the connector config/secrets
Rejected: also route the scheduled-for-deletion flag through an activity | the in-process read is benign — it only flips on delete, and RunTerminateWorkflows reaps those workflows anyway
Rejected: cache the flag once per ContinueAsNew segment | widened the orphan-creation window during uninstall for no determinism gain
Directive: polling period must be read via the activity (recorded in history); reading GetConfig in the workflow body is non-deterministic if the period changes mid-flight
Confidence: high
Scope-risk: moderate
Not-tested: legacy DefaultVersion branch (not exercisable via TestWorkflowEnvironment); orphan schedules from a concurrent fetch racing uninstall (pre-existing; teardown-hardening follow-up)
---------
Co-authored-by: fabrice guery <[email protected]>
37 lines
1.5 KiB
Go
37 lines
1.5 KiB
Go
package workflow
|
|
|
|
import "go.temporal.io/sdk/workflow"
|
|
|
|
const (
|
|
versionFlagOutboxPatternEnabled = "event_outbox_pattern_enabled"
|
|
versionFlagRunNextTaskAsActivity = "run_next_task_as_activity"
|
|
versionFlagPaymentInitiationUpdateAsActivity = "storage_payment_initiation_update_as_activity"
|
|
versionFlagConnectorIDSearchAttributeEnabled = "connector_id_search_attribute_enabled"
|
|
versionFlagDeterministicPollingPeriod = "deterministic_polling_period"
|
|
)
|
|
|
|
func IsEventOutboxPatternEnabled(ctx workflow.Context) bool {
|
|
version := workflow.GetVersion(ctx, versionFlagOutboxPatternEnabled, workflow.DefaultVersion, 1)
|
|
return version > workflow.DefaultVersion
|
|
}
|
|
|
|
func IsRunNextTaskOptimizationsEnabled(ctx workflow.Context) bool {
|
|
version := workflow.GetVersion(ctx, versionFlagRunNextTaskAsActivity, workflow.DefaultVersion, 1)
|
|
return version > workflow.DefaultVersion
|
|
}
|
|
|
|
func IsPaymentInitiationUpdateOptimizationsEnabled(ctx workflow.Context) bool {
|
|
version := workflow.GetVersion(ctx, versionFlagPaymentInitiationUpdateAsActivity, workflow.DefaultVersion, 1)
|
|
return version > workflow.DefaultVersion
|
|
}
|
|
|
|
func IsConnectorIDSearchAttributeEnabled(ctx workflow.Context) bool {
|
|
version := workflow.GetVersion(ctx, versionFlagConnectorIDSearchAttributeEnabled, workflow.DefaultVersion, 1)
|
|
return version > workflow.DefaultVersion
|
|
}
|
|
|
|
func IsDeterministicPollingPeriodEnabled(ctx workflow.Context) bool {
|
|
version := workflow.GetVersion(ctx, versionFlagDeterministicPollingPeriod, workflow.DefaultVersion, 1)
|
|
return version > workflow.DefaultVersion
|
|
}
|