From ead8e68c2df5269f71ba81f427e49e47bdddfacc Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Fri, 18 Sep 2026 22:05:58 -0700 Subject: [PATCH] Enable ASYNC Step input snapshots by default. Keeping this history is the default Server behavior. Operators can still turn it off to skip the extra Blob Store cost. --- docs/content/production/server-operations.mdx | 17 ++++--- docs/design/plan/design-web.md | 8 +-- .../current/production/server-operations.mdx | 14 +++--- protos/README.md | 7 +-- server/config/attribute_store_test.go | 8 +-- server/config/config.go | 14 ++++-- server/config/config_template.yaml | 2 +- server/config/development.yaml | 2 +- server/config/development_cadence.yaml | 2 +- server/integ/config.go | 2 +- server/integ/web_api_test.go | 49 +++++++++++++------ .../async_step_input_snapshot_populator.go | 4 +- server/service/interpreter/activityImpl.go | 2 +- web/app/flows/details/EventDetails.test.tsx | 2 +- web/lib/unavailable.ts | 2 +- 15 files changed, 84 insertions(+), 51 deletions(-) diff --git a/docs/content/production/server-operations.mdx b/docs/content/production/server-operations.mdx index 7479eb00b..b71be1dc3 100644 --- a/docs/content/production/server-operations.mdx +++ b/docs/content/production/server-operations.mdx @@ -124,7 +124,7 @@ the payload size, not the resulting reference size. blobStore: thresholdInBytes: 100 objectIdLength: 10 - asyncStepInputSnapshotsEnabled: false + asyncStepInputSnapshotsEnabled: true supportedStorages: - status: active storageId: p1 @@ -195,13 +195,14 @@ in lexical pages, decodes the escaped Flow and optional Run, and deletes the whole prefix only after the execution no longer exists. Continue listing with the returned continuation token; no manifest is involved. -Async Step input snapshots are disabled by default. Set -**blobStore.asyncStepInputSnapshotsEnabled** to true only when semantic history -must retain the exact inputs sent to successful ASYNC local Step methods. The -snapshot is independent of the payload offload threshold and is not required -for Flow execution, retry, or recovery. When disabled, ASYNC local completion -events report their input as unavailable. SYNC methods and ASYNC methods that -fall back to a regular Activity continue to obtain input from backend history. +Async Step input snapshots are enabled by default when Blob Store is enabled. +They retain the exact inputs sent to successful ASYNC local Step methods so +semantic history can show them. The snapshot is independent of the payload +offload threshold and is not required for Flow execution, retry, or recovery. +Set **blobStore.asyncStepInputSnapshotsEnabled** to false to skip that optional +history cost. When disabled, ASYNC local completion events report their input as +unavailable. SYNC methods and ASYNC methods that fall back to a regular Activity +continue to obtain input from backend history. The optional **blobStore.blobCache** caches S3-backed Attribute objects. A non-empty **directory** enables it and must belong to one Server process. Its diff --git a/docs/design/plan/design-web.md b/docs/design/plan/design-web.md index 71e2ebc8c..4d8a3e91b 100644 --- a/docs/design/plan/design-web.md +++ b/docs/design/plan/design-web.md @@ -375,12 +375,12 @@ Web 只消费统一的 `input/output/context`，不根据 durability 选择额 | ASYNC local failure + regular fallback | fallback ActivityTaskScheduled input | scheduled event metadata；durability 仍为 ASYNC | | ASYNC local failure + budget exhausted | unavailable | LocalActivity failure marker metadata | -`blobStore.asyncStepInputSnapshotsEnabled` 默认关闭。只有明确开启后，成功的 ASYNC -local activity 才保存 snapshot。配置关闭、local snapshot 不存在、external storage 未启用 +`blobStore.asyncStepInputSnapshotsEnabled` 默认开启。Blob Store 启用且未显式关闭时，成功的 ASYNC +local activity 会保存 snapshot。配置关闭、local snapshot 不存在、external storage 未启用 或数据已清理时，server 返回 `input.unavailable=true`。这只代表 step method input snapshot 不可恢复，不代表其中某个 独立 Value blob 加载失败。Web 将它显示为整个 step method input snapshot unavailable， -将默认关闭及 `blobStore.asyncStepInputSnapshotsEnabled` 列为首要可能原因，并明确区分 +将 `blobStore.asyncStepInputSnapshotsEnabled` 关闭及 Blob Store 未启用列为可能原因，并明确区分 单个 Value blob load failure。Web 不显示 page-level data warning；terminal ASYNC failure 说明 short retry budget 可在 sync fallback 前耗尽，因此没有记录 invocation input snapshot，并引导用户沿 Timeline source link 回看调度来源。 @@ -735,7 +735,7 @@ Phase 2 使用 `server/integ/`： - Temporal/Cadence × SYNC/ASYNC：WaitFor/Execute 显示调用时 step input、attributes 和 condition results。 - SYNC scheduled input 和 ASYNC snapshot 都映射为完全相同的 `input/output/context` shape。 - regular Activity input proto 保持不变；第二个 activity argument 为 null 时 Temporal/Cadence 都能解码。 -- 开启 `blobStore.asyncStepInputSnapshotsEnabled` 后，ASYNC local success 保存 `InternalAsyncStepInputSnapshot`；marker 中不增加完整 request。 +- 默认开启 `blobStore.asyncStepInputSnapshotsEnabled` 后，ASYNC local success 保存 `InternalAsyncStepInputSnapshot`；marker 中不增加完整 request。关闭该选项可跳过这项可选 history 成本。 - method options：SYNC 从 scheduled metadata 转换；ASYNC success 从 snapshot 恢复，fallback 从 local failure metadata 恢复。 - channel values、多个 timers、ANY/ALL results 从保存的 worker request 精确恢复。 - local failure fallback 使用 regular Activity history request，且不暴露 local failure。 diff --git a/docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/production/server-operations.mdx b/docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/production/server-operations.mdx index 965e9a6ff..b6200dfc6 100644 --- a/docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/production/server-operations.mdx +++ b/docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/production/server-operations.mdx @@ -109,7 +109,7 @@ object 会被 offload。可通过 **blobStore.thresholdInBytes** 覆盖默认值 blobStore: thresholdInBytes: 100 objectIdLength: 10 - asyncStepInputSnapshotsEnabled: false + asyncStepInputSnapshotsEnabled: true supportedStorages: - status: active storageId: p1 @@ -169,12 +169,12 @@ cleanup 保留每个 Flow prefix 开头的 UTC date。它按字典序分页列 Flow 和可选 Run，并且只在 execution 已不存在时删除整个 prefix。随后使用返回的 continuation token 继续 list；该过程不使用 manifest。 -ASYNC Step input snapshot 默认关闭。只有当 semantic history 必须保留成功的 ASYNC local -Step method 收到的准确 input 时，才将 -**blobStore.asyncStepInputSnapshotsEnabled** 设置为 true。snapshot 不受 payload offload -threshold 影响，也不是 Flow execution、retry 或 recovery 所必需。关闭后，ASYNC local -completion event 会将 input 标记为 unavailable。SYNC method 和 fallback 到 regular -Activity 的 ASYNC method 仍会从 backend history 获得 input。 +ASYNC Step input snapshot 在 Blob Store 启用时默认开启。它会保留成功的 ASYNC local +Step method 收到的准确 input，供 semantic history 展示。snapshot 不受 payload offload +threshold 影响，也不是 Flow execution、retry 或 recovery 所必需。将 +**blobStore.asyncStepInputSnapshotsEnabled** 设置为 false 可跳过这项可选的 history +成本。关闭后，ASYNC local completion event 会将 input 标记为 unavailable。SYNC method +和 fallback 到 regular Activity 的 ASYNC method 仍会从 backend history 获得 input。 可选的 **blobStore.blobCache** 缓存 S3 支持的 Attribute object。非空 **directory** 会启用 它，并且只能属于一个 Server process。默认 budget 为 1 GiB。过大或被拒绝的 object 会 diff --git a/protos/README.md b/protos/README.md index b7df3c69b..d0611a9d5 100644 --- a/protos/README.md +++ b/protos/README.md @@ -302,9 +302,10 @@ boundary. A fallback regular activity carries only its prior attempt count and first-attempt time in `Context`. `InternalAsyncStepInputSnapshot` is the run-scoped request and method-options -record. Writing and reading these records requires the opt-in -`blobStore.asyncStepInputSnapshotsEnabled` setting. None of these internal -types is returned by `FlowService`. +record. Writing and reading these records uses the +`blobStore.asyncStepInputSnapshotsEnabled` setting, which defaults to true when +Blob Store is enabled. Set it to false to skip that optional history cost. None +of these internal types is returned by `FlowService`. `LoadBlobs` resolves batches of string/object blob arms. Callers should dedupe by value kind and blob ID before loading. Missing objects and unconfigured store diff --git a/server/config/attribute_store_test.go b/server/config/attribute_store_test.go index cf7d393eb..df4ade8a5 100644 --- a/server/config/attribute_store_test.go +++ b/server/config/attribute_store_test.go @@ -192,7 +192,7 @@ func TestBlobStoreDefaults(t *testing.T) { cfg, err := NewConfig(path) require.NoError(t, err) require.True(t, cfg.BlobStore.EffectiveEnabled()) - require.False(t, cfg.BlobStore.AsyncStepInputSnapshotsEnabled) + require.True(t, cfg.BlobStore.EffectiveAsyncStepInputSnapshotsEnabled()) require.Equal(t, 100, cfg.BlobStore.EffectiveThresholdInBytes()) require.Equal(t, DefaultBlobStoreObjectIDLength, cfg.BlobStore.EffectiveObjectIDLength()) require.Equal(t, 100*time.Millisecond, cfg.AttributeStore.EffectiveSyncRetryPolicy().InitialInterval) @@ -208,11 +208,13 @@ func TestBlobStoreObjectIDLengthValidation(t *testing.T) { func TestBlobStoreAsyncStepInputSnapshotsRequireBlobStore(t *testing.T) { require.NoError(t, (BlobStoreConfig{}).Validate()) - require.NoError(t, (BlobStoreConfig{AsyncStepInputSnapshotsEnabled: true}).Validate()) + require.NoError(t, (BlobStoreConfig{Enabled: ptr.Any(false)}).Validate()) + require.NoError(t, (BlobStoreConfig{AsyncStepInputSnapshotsEnabled: ptr.Any(true)}).Validate()) + require.False(t, (BlobStoreConfig{Enabled: ptr.Any(false)}).EffectiveAsyncStepInputSnapshotsEnabled()) err := (BlobStoreConfig{ Enabled: ptr.Any(false), - AsyncStepInputSnapshotsEnabled: true, + AsyncStepInputSnapshotsEnabled: ptr.Any(true), }).Validate() require.ErrorContains(t, err, "asyncStepInputSnapshotsEnabled") } diff --git a/server/config/config.go b/server/config/config.go index 1931ee4f6..d30d8ad0b 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -212,8 +212,8 @@ type ( BlobStoreConfig struct { // Enabled turns blob offload on or off. Default true when omitted. Enabled *bool `yaml:"enabled"` - // AsyncStepInputSnapshotsEnabled stores successful asynchronous local Step method inputs for semantic history. Default false. Requires Enabled and is immutable after startup. - AsyncStepInputSnapshotsEnabled bool `yaml:"asyncStepInputSnapshotsEnabled"` + // AsyncStepInputSnapshotsEnabled stores successful asynchronous local Step method inputs for semantic history. Default true when omitted and Blob Store is enabled. Requires Enabled when true and is immutable after startup. Set false to skip this optional history cost. + AsyncStepInputSnapshotsEnabled *bool `yaml:"asyncStepInputSnapshotsEnabled"` // LazyLoading turns lazy loading on or off. // When on, server will only send blobIDs to worker for worker APIs(invoke waitFor/execute/RPC) and GetAttribute API. // Worker wil call LoadBlobs API to get the actual values. @@ -758,9 +758,17 @@ func (c BlobStoreConfig) EffectiveObjectIDLength() int { return c.ObjectIDLength } +// EffectiveAsyncStepInputSnapshotsEnabled returns whether successful ASYNC local Step inputs are stored for semantic history. Default true when omitted and Blob Store is enabled. +func (c BlobStoreConfig) EffectiveAsyncStepInputSnapshotsEnabled() bool { + if c.AsyncStepInputSnapshotsEnabled == nil { + return c.EffectiveEnabled() + } + return *c.AsyncStepInputSnapshotsEnabled +} + // Validate checks Blob Store identifier and cache settings. func (c BlobStoreConfig) Validate() error { - if c.AsyncStepInputSnapshotsEnabled && !c.EffectiveEnabled() { + if c.EffectiveAsyncStepInputSnapshotsEnabled() && !c.EffectiveEnabled() { return fmt.Errorf("blobStore asyncStepInputSnapshotsEnabled requires blobStore.enabled") } if c.ObjectIDLength < 0 { diff --git a/server/config/config_template.yaml b/server/config/config_template.yaml index a62c9b69a..32ce32b50 100644 --- a/server/config/config_template.yaml +++ b/server/config/config_template.yaml @@ -53,7 +53,7 @@ interpreter: cloudAPIKey: "" blobStore: enabled: true - asyncStepInputSnapshotsEnabled: false + asyncStepInputSnapshotsEnabled: true thresholdInBytes: 100 objectIdLength: 10 historyRetentionInDays: 3 diff --git a/server/config/development.yaml b/server/config/development.yaml index 1b078acb1..de8391d1c 100644 --- a/server/config/development.yaml +++ b/server/config/development.yaml @@ -43,7 +43,7 @@ interpreter: timerType: histogram blobStore: enabled: true - asyncStepInputSnapshotsEnabled: false + asyncStepInputSnapshotsEnabled: true thresholdInBytes: 100 objectIdLength: 10 historyRetentionInDays: 3 diff --git a/server/config/development_cadence.yaml b/server/config/development_cadence.yaml index f4e2ba44b..1d4080628 100644 --- a/server/config/development_cadence.yaml +++ b/server/config/development_cadence.yaml @@ -39,7 +39,7 @@ interpreter: adminSecurityToken: "" blobStore: enabled: true - asyncStepInputSnapshotsEnabled: false + asyncStepInputSnapshotsEnabled: true thresholdInBytes: 100 objectIdLength: 10 historyRetentionInDays: 3 diff --git a/server/integ/config.go b/server/integ/config.go index 6530cb13b..465efd6ca 100644 --- a/server/integ/config.go +++ b/server/integ/config.go @@ -35,7 +35,7 @@ type DexServiceTestConfig struct { StreamStore config.StreamStoreConfig BlobCacheDirectory string BlobStoreEnabled *bool - AsyncStepInputSnapshotsEnabled bool + AsyncStepInputSnapshotsEnabled *bool IncludeRPCInputOutputIntoHistory bool UseTemporalSynchronousUpdateForAllRPCs bool TemporalMetricsHandler client.MetricsHandler diff --git a/server/integ/web_api_test.go b/server/integ/web_api_test.go index 542a04bda..fa6281bab 100644 --- a/server/integ/web_api_test.go +++ b/server/integ/web_api_test.go @@ -86,8 +86,11 @@ func testWebAPI(t *testing.T, backendType service.BackendType) { t.Run("async-local-fallback", func(t *testing.T) { testWebAsyncLocalFallback(t, backendType) }) - t.Run("async-step-input-snapshots-disabled-by-default", func(t *testing.T) { - testWebAsyncStepInputSnapshotsDisabledByDefault(t, backendType) + t.Run("async-step-input-snapshots-enabled-by-default", func(t *testing.T) { + testWebAsyncStepInputSnapshotsEnabledByDefault(t, backendType) + }) + t.Run("async-step-input-snapshots-can-be-disabled", func(t *testing.T) { + testWebAsyncStepInputSnapshotsCanBeDisabled(t, backendType) }) t.Run("time-travel-snapshot-origin", func(t *testing.T) { testWebTimeTravelSnapshotOrigin(t, backendType) @@ -136,7 +139,7 @@ func testWebTimeTravelSnapshotOrigin(t *testing.T, backendType service.BackendTy runtime := startDexService(t, DexServiceTestConfig{ BackendType: backendType, LocalBlobDirectory: t.TempDir(), - AsyncStepInputSnapshotsEnabled: true, + AsyncStepInputSnapshotsEnabled: ptr.Any(true), }) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() @@ -958,18 +961,36 @@ func testWebStepInputWithoutStorage( require.Equal(t, expectedUnavailable, executeInput.GetUnavailable()) } -func testWebAsyncStepInputSnapshotsDisabledByDefault( +func testWebAsyncStepInputSnapshotsEnabledByDefault( + t *testing.T, + backendType service.BackendType, +) { + assertWebAsyncStepInputSnapshotAvailability(t, backendType, nil, true) +} + +func testWebAsyncStepInputSnapshotsCanBeDisabled( + t *testing.T, + backendType service.BackendType, +) { + assertWebAsyncStepInputSnapshotAvailability(t, backendType, ptr.Any(false), false) +} + +func assertWebAsyncStepInputSnapshotAvailability( t *testing.T, backendType service.BackendType, + snapshotsEnabled *bool, + expectRecorded bool, ) { + t.Helper() workerTarget := startWorker(t, basic.NewHandler()) runtime := startDexService(t, DexServiceTestConfig{ - BackendType: backendType, - LocalBlobDirectory: t.TempDir(), + BackendType: backendType, + LocalBlobDirectory: t.TempDir(), + AsyncStepInputSnapshotsEnabled: snapshotsEnabled, }) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - flowID := "web-async-snapshots-disabled-" + uuid.NewString() + flowID := "web-async-snapshots-" + uuid.NewString() startResponse, err := runtime.FlowClient.StartFlow(ctx, &dexpb.StartFlowRequest{ RequestId: newRequestID(), FlowId: flowID, @@ -991,8 +1012,8 @@ func testWebAsyncStepInputSnapshotsDisabledByDefault( executeEvent := firstExecuteEvent(events) require.NotNil(t, waitForEvent) require.NotNil(t, executeEvent) - require.True(t, waitForEvent.GetInput().GetUnavailable()) - require.True(t, executeEvent.GetInput().GetUnavailable()) + require.Equal(t, !expectRecorded, waitForEvent.GetInput().GetUnavailable()) + require.Equal(t, !expectRecorded, executeEvent.GetInput().GetUnavailable()) description, err := runtime.UnifiedClient.DescribeWorkflowExecution( ctx, @@ -1014,7 +1035,7 @@ func testWebAsyncStepInputSnapshotsDisabledByDefault( method, ) require.NoError(t, readErr) - require.False(t, found) + require.Equal(t, expectRecorded, found) } } @@ -1025,7 +1046,7 @@ func testWebParallelAttributeSnapshots(t *testing.T, backendType service.Backend BackendType: backendType, LocalBlobDirectory: t.TempDir(), LocalBlobThreshold: 10, - AsyncStepInputSnapshotsEnabled: true, + AsyncStepInputSnapshotsEnabled: ptr.Any(true), }) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -1217,7 +1238,7 @@ func testWebConditionResults( LazyLoading: ptr.Any(true), LocalBlobDirectory: t.TempDir(), LocalBlobThreshold: 10, - AsyncStepInputSnapshotsEnabled: true, + AsyncStepInputSnapshotsEnabled: ptr.Any(true), }) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -1446,7 +1467,7 @@ func testWebHistoryAndSummary( LazyLoading: ptr.Any(lazyLoading), LocalBlobDirectory: blobDirectory, LocalBlobThreshold: 10, - AsyncStepInputSnapshotsEnabled: true, + AsyncStepInputSnapshotsEnabled: ptr.Any(true), }) ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) defer cancel() @@ -1707,7 +1728,7 @@ func testWebCurrentState(t *testing.T, backendType service.BackendType, lazyLoad LazyLoading: ptr.Any(lazyLoading), LocalBlobDirectory: t.TempDir(), LocalBlobThreshold: 10, - AsyncStepInputSnapshotsEnabled: true, + AsyncStepInputSnapshotsEnabled: ptr.Any(true), }) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() diff --git a/server/service/client/history/async_step_input_snapshot_populator.go b/server/service/client/history/async_step_input_snapshot_populator.go index 250eff143..f610b9ead 100644 --- a/server/service/client/history/async_step_input_snapshot_populator.go +++ b/server/service/client/history/async_step_input_snapshot_populator.go @@ -44,7 +44,7 @@ func NewAsyncStepInputSnapshotPopulator( if cfg == nil || client == nil { panic("async step input snapshot populator requires config and client") } - if cfg.AsyncStepInputSnapshotsEnabled && cfg.EffectiveEnabled() && store == nil { + if cfg.EffectiveAsyncStepInputSnapshotsEnabled() && store == nil { panic("async step input snapshot populator requires a blob store when enabled") } return &AsyncStepInputSnapshotPopulator{ @@ -66,7 +66,7 @@ func (p *AsyncStepInputSnapshotPopulator) Populate( if !hasMissingStepEventInput(events) { return nil } - if !p.cfg.AsyncStepInputSnapshotsEnabled || !p.cfg.EffectiveEnabled() { + if !p.cfg.EffectiveAsyncStepInputSnapshotsEnabled() { markMissingStepEventInputsUnavailable(events) return nil } diff --git a/server/service/interpreter/activityImpl.go b/server/service/interpreter/activityImpl.go index ea63bce52..01dc04760 100644 --- a/server/service/interpreter/activityImpl.go +++ b/server/service/interpreter/activityImpl.go @@ -700,7 +700,7 @@ func (a *Activities) persistStepEventInput( input *dexpb.InternalAsyncStepInputSnapshot, ) error { if !activityInfo.IsLocalActivity || - !a.cfg.BlobStore.AsyncStepInputSnapshotsEnabled || + !a.cfg.BlobStore.EffectiveAsyncStepInputSnapshotsEnabled() || !a.cfg.BlobStore.EffectiveEnabled() { return nil } diff --git a/web/app/flows/details/EventDetails.test.tsx b/web/app/flows/details/EventDetails.test.tsx index 321697afe..a91a1102b 100644 --- a/web/app/flows/details/EventDetails.test.tsx +++ b/web/app/flows/details/EventDetails.test.tsx @@ -240,8 +240,8 @@ describe('selected step event details', () => { expect(markup).toContain('Step method input snapshot unavailable'); expect(markup).toContain(STEP_INPUT_SNAPSHOT_UNAVAILABLE); - expect(markup).toContain('ASYNC Step input snapshots are disabled by default'); expect(markup).toContain('blobStore.asyncStepInputSnapshotsEnabled'); + expect(markup).toContain('set blobStore.asyncStepInputSnapshotsEnabled to false'); expect(markup).toContain('Blob Store may be disabled'); expect(markup).toContain('full invocation input snapshot'); expect(markup).toContain('does not indicate an individual Value blob load failure'); diff --git a/web/lib/unavailable.ts b/web/lib/unavailable.ts index 0b76a0a8e..c256eb63a 100644 --- a/web/lib/unavailable.ts +++ b/web/lib/unavailable.ts @@ -10,4 +10,4 @@ export const VALUE_BLOB_UNAVAILABLE = 'Value blob unavailable'; export const ASYNC_STEP_INPUT_SNAPSHOT_NOT_RECORDED = 'This ASYNC Step method exhausted its short retry policy before sync fallback, so its full invocation input snapshot was not recorded. This does not indicate an individual Value blob load failure.'; export const STEP_INPUT_SNAPSHOT_UNAVAILABLE = - 'ASYNC Step input snapshots are disabled by default. The Server may not have enabled blobStore.asyncStepInputSnapshotsEnabled, Blob Store may be disabled, or this snapshot may no longer be retained. The full invocation input snapshot is unavailable; this does not indicate an individual Value blob load failure.'; + 'The Server may have set blobStore.asyncStepInputSnapshotsEnabled to false, Blob Store may be disabled, or this snapshot may no longer be retained. The full invocation input snapshot is unavailable; this does not indicate an individual Value blob load failure.';