Long-Running Workflows

ธุรกรรมส่วนใหญ่จบใน 30 วินาที แต่ระบบ wallet จริงมีงานที่กินเวลาเป็นวัน เป็นเดือน หรือไม่จบเลย — รอ settlement T+1, กระทบยอดตี 2 ทุกคืน, บัญชีลูกค้า 1 ใบที่มีชีวิตอยู่ตลอด บทนี้ว่าด้วยเครื่องมือที่ทำให้ workflow อยู่ได้นานโดยไม่ระเบิด

Timer ที่รอดจาก worker restart#

time.Sleep ใน workflow คือ non-determinism ตามที่ว่าไว้ในบทที่ 4 มันบล็อก goroutine จริงและหายไปพร้อม worker ที่ตาย

workflow.Sleep ไม่ใช่การหลับ — มันคือคำสั่งให้ server ตั้งนาฬิกาไว้ให้ worker ปล่อย workflow ออกจาก memory ได้เลย พอถึงเวลา server จะ schedule task ใหม่ แล้ว replay กลับมาที่บรรทัดถัดไป ระหว่างที่รอไม่มี process ไหนของเราค้างอยู่เลย

// รอครบ T+1 แล้วค่อยตรวจว่าเงินเข้าบัญชีปลายทางจริงหรือยัง
func SettlementCheckWorkflow(ctx workflow.Context, ref BankTransferRef) error {
	if err := workflow.Sleep(ctx, 24*time.Hour); err != nil {
		return err // ctx ถูก cancel ระหว่างรอ
	}

	actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 60 * time.Second,
	})
	var a *activities.Activities

	var settled bool
	if err := workflow.ExecuteActivity(actCtx, a.CheckSettlement, ref).Get(actCtx, &settled); err != nil {
		return err
	}
	if !settled {
		return workflow.ExecuteActivity(actCtx, a.RaiseOpsAlert, ref.TransactionID,
			"not settled after T+1").Get(actCtx, nil)
	}
	return nil
}

รอเป็นเดือนก็ได้ ค่าใช้จ่ายเท่ากับ event 2 ตัวใน history (TimerStarted, TimerFired)

เลือกให้ถูกแบบ:

// รอ ops อนุมัติภายใน 8 ชั่วโมง — สั้นกว่าเขียน Selector เองเมื่อเงื่อนไขเป็น bool ธรรมดา
approved := false
workflow.Go(ctx, func(gctx workflow.Context) {
	workflow.GetSignalChannel(gctx, "ops-approved").Receive(gctx, &approved)
})

ok, err := workflow.AwaitWithTimeout(ctx, 8*time.Hour, func() bool { return approved })
if err != nil {
	return err
}
if !ok {
	return temporal.NewNonRetryableApplicationError(
		"approval timed out", "ApprovalTimeout", nil)
}

Selector — จัดการเหตุการณ์ที่แข่งกัน#

บทที่ 7 ใช้ Selector รอ "OTP หรือ timeout อันไหนมาก่อน" มาแล้ว ตรงนี้ขยายไปที่กรณีซับซ้อนขึ้น: รอ callback จากธนาคาร แต่ระหว่างรอก็ poll สถานะเป็นระยะ และยอมให้ ops สั่งยกเลิกได้ตลอด

func AwaitBankResultWorkflow(ctx workflow.Context, ref BankTransferRef) (string, error) {
	actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 30 * time.Second,
	})
	var a *activities.Activities

	callbackCh := workflow.GetSignalChannel(ctx, "bank-callback")
	abortCh := workflow.GetSignalChannel(ctx, "ops-abort")

	deadline := workflow.NewTimer(ctx, 6*time.Hour)
	outcome := ""

	for outcome == "" {
		// poll timer ต้องสร้างใหม่ทุกรอบ — Future 1 ตัวยิงได้ครั้งเดียว
		poll := workflow.NewTimer(ctx, 5*time.Minute)

		sel := workflow.NewSelector(ctx)
		sel.AddReceive(callbackCh, func(c workflow.ReceiveChannel, _ bool) {
			var cb BankCallback
			c.Receive(ctx, &cb)
			outcome = cb.Status
		})
		sel.AddReceive(abortCh, func(c workflow.ReceiveChannel, _ bool) {
			c.Receive(ctx, nil)
			outcome = "aborted"
		})
		sel.AddFuture(poll, func(f workflow.Future) {
			if f.Get(ctx, nil) != nil {
				return // timer ถูก cancel
			}
			var status string
			if err := workflow.ExecuteActivity(actCtx, a.PollBankStatus, ref).Get(actCtx, &status); err == nil &&
				status != "pending" {
				outcome = status
			}
		})
		sel.AddFuture(deadline, func(f workflow.Future) {
			if f.Get(ctx, nil) == nil {
				outcome = "timeout"
			}
		})

		sel.Select(ctx)
	}

	return outcome, nil
}

จุดที่พลาดกันบ่อย:

Child Workflow#

Child Workflow คือ workflow ที่ถูกสั่งจาก workflow อีกตัว มี history ของตัวเอง มี retry policy ของตัวเอง และล้มได้โดยไม่ลาก parent ไปด้วย

ใช้เมื่อ:

• กัน history บวม — parent สั่งงาน 500 รายการ ถ้าทำในตัวเองจะได้ event หลายพัน
• แยก failure domain — ธนาคาร A ล่ม ไม่ควรทำให้รอบกระทบยอดของธนาคาร B พัง
• คนละ retry / timeout policy — งาน batch ยอมช้าได้ งาน real-time ยอมไม่ได้
• อายุไม่เท่ากัน — parent จบแล้วแต่ลูกต้องอยู่ต่อ (ABANDON)

เคสจริง: รอบกระทบยอดรายวัน แตกเป็น 1 child ต่อ 1 ธนาคาร

System map

func DailyReconciliationWorkflow(ctx workflow.Context, day string) (ReconResult, error) {
	banks := []string{"SCB", "KBANK", "BBL"}

	// ยิงพร้อมกันทั้งหมดก่อน แล้วค่อยเก็บผล — ไม่งั้นจะกลายเป็นทำทีละธนาคาร
	futures := make(map[string]workflow.ChildWorkflowFuture, len(banks))
	for _, bank := range banks {
		bankCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
			// ตั้ง ID เองเสมอ เพื่อให้ ops ค้นเจอใน Web UI และกันรันซ้ำรอบเดิม
			WorkflowID:               fmt.Sprintf("recon-%s-%s", day, bank),
			TaskQueue:                "recon",
			WorkflowExecutionTimeout: 4 * time.Hour,
			RetryPolicy: &temporal.RetryPolicy{
				MaximumAttempts: 3,
			},
		})
		futures[bank] = workflow.ExecuteChildWorkflow(bankCtx, ReconcileBankWorkflow, day, bank)
	}

	result := ReconResult{Day: day, PerBank: map[string]BankReconResult{}}
	for bank, f := range futures {
		var r BankReconResult
		if err := f.Get(ctx, &r); err != nil {
			// ธนาคารเดียวพัง ไม่ล้มทั้งรอบ — บันทึกไว้แล้วไปต่อ
			workflow.GetLogger(ctx).Error("bank recon failed", "bank", bank, "error", err)
			result.Failed = append(result.Failed, bank)
			continue
		}
		result.PerBank[bank] = r
		result.MismatchSatang += r.MismatchSatang
	}

	return result, nil
}

Options ที่ต้องรู้#

ParentClosePolicy มี 3 ค่า:

ยิงแล้วไม่รอ#

ถ้าอยากให้ child อยู่ต่อหลัง parent จบ ต้องรอให้มัน เริ่ม จริงก่อน ไม่งั้น parent อาจจบก่อนที่คำสั่งสร้าง child จะถูกส่งออกไป

detachedCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
	WorkflowID:        "settlement-check-" + ref.TransactionID,
	ParentClosePolicy: enumspb.PARENT_CLOSE_POLICY_ABANDON,
})

child := workflow.ExecuteChildWorkflow(detachedCtx, SettlementCheckWorkflow, ref)

// รอแค่ "เริ่มแล้ว" ไม่ต้องรอผลลัพธ์
var exec workflow.Execution
if err := child.GetChildWorkflowExecution().Get(ctx, &exec); err != nil {
	return err
}
workflow.GetLogger(ctx).Info("settlement check started", "runID", exec.RunID)

ส่ง signal เข้า child ได้ตรงๆ ผ่าน future เดียวกัน:

err := child.SignalChildWorkflow(ctx, "bank-callback", callbackPayload).Get(ctx, nil)

Continue-As-New#

Event History โตขึ้นเรื่อยๆ ตามจำนวนสิ่งที่ workflow ทำ และ server มีเพดานทั้งจำนวน event และขนาดรวม พอชนเพดาน workflow จะถูกบังคับให้ล้ม — สำหรับ workflow ที่วนลูปไม่รู้จบ นี่ไม่ใช่ "ถ้าเกิด" แต่คือ "เมื่อไหร่"

Continue-As-New คือการปิด run ปัจจุบันแล้วเปิด run ใหม่ทันที ด้วย Workflow ID เดิม แต่ history ว่างเปล่า สถานะที่อยากเก็บต้องส่งต่อไปเป็น input

System map

type WalletState struct {
	WalletID      string
	BalanceSatang int64
	// เก็บเท่าที่จำเป็นต่อการทำงานรอบถัดไป — ห้ามสะสมประวัติทั้งหมดไว้ตรงนี้
	LastTxnID     string
	DailyUsed     int64
	DailyResetAt  time.Time
}

func WalletAccountWorkflow(ctx workflow.Context, state WalletState) error {
	// ... ทำงานเป็นลูป ...

	if workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
		return workflow.NewContinueAsNewError(ctx, WalletAccountWorkflow, state)
	}
	return nil
}

GetContinueAsNewSuggested() ให้ server เป็นคนบอกว่าใกล้เพดานแล้ว ซึ่งดีกว่า hardcode ตัวเลข เพราะเพดานเป็นค่า config ระดับ cluster ที่เปลี่ยนได้ ถ้าต้องการเกณฑ์ของตัวเองเพิ่ม (เช่นอยากตัดถี่กว่านั้นเพื่อให้ replay เร็ว) ใช้ GetCurrentHistoryLength() ประกอบ:

func shouldContinueAsNew(ctx workflow.Context) bool {
	info := workflow.GetInfo(ctx)
	if info.GetContinueAsNewSuggested() {
		return true
	}
	return info.GetCurrentHistoryLength() > 2_000
}

Drain ให้ถูก:

func drainAndContinue(ctx workflow.Context, state WalletState, ch workflow.ReceiveChannel) error {
	// 1. รอ signal/update handler ที่ยังทำงานค้างให้เสร็จก่อน (TMPRL1102)
	if err := workflow.Await(ctx, func() bool {
		return workflow.AllHandlersFinished(ctx)
	}); err != nil {
		return err
	}

	// 2. ดูดข้อความที่ค้างในคิวออกให้หมด แล้วประมวลผลตามปกติ
	for {
		var cmd WalletCommand
		if !ch.ReceiveAsync(&cmd) {
			break
		}
		state = applyCommand(state, cmd)
	}

	// 3. ค่อย continue
	return workflow.NewContinueAsNewError(ctx, WalletAccountWorkflow, state)
}

Entity Workflow#

Entity Workflow คือการมองว่า "1 บัญชี = 1 workflow ที่ไม่จบ" แทนที่จะเปิด workflow ใหม่ทุกครั้งที่มีธุรกรรม เราเปิดตัวเดียวต่อ wallet แล้วยิงคำสั่งเข้าไปด้วย signal/update ตลอดอายุการใช้งาน

ประโยชน์ในบริบทการเงินคือได้ serialization ฟรี — คำสั่งทั้งหมดของ wallet ใบนั้น ถูกประมวลผลทีละคำสั่งตามลำดับใน workflow เดียว ไม่มี race, ไม่ต้องล็อก DB, และตรวจ daily limit ได้จาก state ในหน่วยความจำโดยตรง

func WalletAccountWorkflow(ctx workflow.Context, state WalletState) error {
	logger := workflow.GetLogger(ctx)

	// อ่านสถานะได้ตลอดจากหน้าจอ ops
	if err := workflow.SetQueryHandler(ctx, "balance", func() (int64, error) {
		return state.BalanceSatang, nil
	}); err != nil {
		return err
	}

	// สั่งเติมเงินแล้วรู้ผลทันที — คู่กับ Update-with-Start จากบทที่ 7
	err := workflow.SetUpdateHandlerWithOptions(ctx, "top-up",
		func(ctx workflow.Context, cmd TopUpCommand) (TopUpAccepted, error) {
			// idempotency: คำสั่งเดิมส่งซ้ำต้องได้ผลเดิม ไม่ใช่เติม 2 รอบ
			if cmd.TransactionID == state.LastTxnID {
				return TopUpAccepted{BalanceSatang: state.BalanceSatang, Duplicate: true}, nil
			}

			actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
				StartToCloseTimeout: 60 * time.Second,
			})
			var a *activities.Activities
			var bankRef string
			if err := workflow.ExecuteActivity(actCtx, a.ChargeBank, cmd).Get(actCtx, &bankRef); err != nil {
				return TopUpAccepted{}, err
			}

			state.BalanceSatang += cmd.AmountSatang
			state.DailyUsed += cmd.AmountSatang
			state.LastTxnID = cmd.TransactionID
			return TopUpAccepted{BalanceSatang: state.BalanceSatang, BankRef: bankRef}, nil
		},
		workflow.UpdateHandlerOptions{
			Validator: func(ctx workflow.Context, cmd TopUpCommand) error {
				if cmd.AmountSatang <= 0 {
					return fmt.Errorf("amount must be positive")
				}
				if state.DailyUsed+cmd.AmountSatang > 100_000_00 {
					return fmt.Errorf("daily limit exceeded")
				}
				return nil
			},
		},
	)
	if err != nil {
		return err
	}

	closeCh := workflow.GetSignalChannel(ctx, "close-account")
	closed := false
	workflow.Go(ctx, func(gctx workflow.Context) {
		closeCh.Receive(gctx, nil)
		closed = true
	})

	// รีเซ็ตยอดใช้รายวัน และตรวจว่าถึงเวลา continue-as-new หรือยัง
	for {
		nextReset := workflow.Now(ctx).Truncate(24 * time.Hour).Add(24 * time.Hour)
		fired, err := workflow.AwaitWithTimeout(ctx, nextReset.Sub(workflow.Now(ctx)),
			func() bool { return closed || workflow.GetInfo(ctx).GetContinueAsNewSuggested() })
		if err != nil {
			return err
		}

		if !fired {
			state.DailyUsed = 0
			state.DailyResetAt = workflow.Now(ctx)
			continue
		}
		if closed {
			logger.Info("wallet closed", "walletID", state.WalletID)
			return workflow.Await(ctx, func() bool { return workflow.AllHandlersFinished(ctx) })
		}

		if err := workflow.Await(ctx, func() bool {
			return workflow.AllHandlersFinished(ctx)
		}); err != nil {
			return err
		}
		return workflow.NewContinueAsNewError(ctx, WalletAccountWorkflow, state)
	}
}

ฝั่ง API ใช้ Update-with-Start เพื่อ "เปิดถ้ายังไม่มี แล้วสั่งเลย" ตามที่อธิบายไว้ใน บทที่ 7 จึงไม่ต้องมีขั้นตอน provision บัญชีแยกต่างหาก

Schedule — งานที่ต้องรันตามเวลา#

อย่าเขียน for { workflow.Sleep(24h); doWork() } เพื่อทำงานประจำวัน มันดูง่ายแต่ไม่มี UI ให้ ops กด pause, ย้อนหลังไม่ได้, และถ้ารอบก่อนยังไม่จบก็ไม่มีใครห้าม

Schedule เป็น object ระดับ server ที่จัดการเรื่องพวกนี้ให้:

handle, err := c.ScheduleClient().Create(context.Background(), client.ScheduleOptions{
	ID: "daily-reconciliation",
	Spec: client.ScheduleSpec{
		Calendars: []client.ScheduleCalendarSpec{
			{
				Hour:    []client.ScheduleRange{{Start: 2}},
				Minute:  []client.ScheduleRange{{Start: 0}},
				Comment: "กระทบยอดรายวันหลังปิดรอบธนาคาร",
			},
		},
		TimeZoneName: "Asia/Bangkok", // สำคัญ — default คือ UTC
		Jitter:       5 * time.Minute,
	},
	Action: &client.ScheduleWorkflowAction{
		ID:                       "recon", // จะถูกต่อท้ายด้วย timestamp ของแต่ละรอบ
		Workflow:                 workflows.DailyReconciliationWorkflow,
		TaskQueue:                "recon",
		WorkflowExecutionTimeout: 4 * time.Hour,
	},
	Overlap:        enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
	CatchupWindow:  2 * time.Hour,
	PauseOnFailure: true,
})

Overlap Policy — ข้อที่สำคัญที่สุดสำหรับงานการเงิน#

ถ้ารอบเมื่อวานยังไม่จบแล้วถึงเวลารอบใหม่ จะทำยังไง

PauseOnFailure: true ทำให้ schedule หยุดตัวเองเมื่อรอบหนึ่งล้ม คุ้มค่ามากกับงานกระทบยอด เพราะรอบถัดไปที่รันบนฐานข้อมูลที่ยังไม่ถูกแก้ก็จะล้มซ้ำอยู่ดี แถมยังกลบร่องรอยของรอบแรก

จัดการ schedule#

h := c.ScheduleClient().GetHandle(context.Background(), "daily-reconciliation")

_ = h.Pause(ctx, client.SchedulePauseOptions{Note: "ปิดระบบบำรุงรักษา"})
_ = h.Unpause(ctx, client.ScheduleUnpauseOptions{Note: "เสร็จแล้ว"})

// สั่งรันเดี๋ยวนี้เลย โดยไม่รบกวนตารางปกติ
_ = h.Trigger(ctx, client.ScheduleTriggerOptions{})

// ย้อนรันรอบที่ขาดไป เช่นตอน worker ล่มไป 3 วัน
_ = h.Backfill(ctx, client.ScheduleBackfillOptions{
	Backfill: []client.ScheduleBackfill{{
		Start:   time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC),
		End:     time.Date(2026, 8, 3, 0, 0, 0, 0, time.UTC),
		Overlap: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
	}},
})

desc, _ := h.Describe(ctx)

จาก CLI:

temporal schedule create \
  --schedule-id daily-reconciliation \
  --calendar '{"hour":"2","minute":"0"}' \
  --workflow-id recon \
  --task-queue recon \
  --type DailyReconciliationWorkflow \
  --overlap-policy Skip

temporal schedule describe --schedule-id daily-reconciliation
temporal schedule toggle   --schedule-id daily-reconciliation --pause --reason "maintenance"
temporal schedule trigger  --schedule-id daily-reconciliation
temporal schedule backfill --schedule-id daily-reconciliation \
  --start-time 2026-08-01T00:00:00Z \
  --end-time   2026-08-03T00:00:00Z \
  --overlap-policy BufferAll

เลือกเครื่องมือให้ถูกงาน#