Documentation
¶
Overview ¶
Package jobstatus は、非同期ジョブの進行状況をリモートストレージ上の JSON として 読み書きします。
生成の成否がこれまで Slack 通知にしか残らず、失敗したジョブが UI から消えていた 問題を解消するための記録層です。あわせて、Cloud Tasks の at-least-once 配信に対する 再実行ガードの根拠にもなります。
Index ¶
- Variables
- type Carrier
- type Locator
- type Recorder
- func (r *Recorder[T]) AlreadySucceeded(ctx context.Context, jobID string) (bool, error)
- func (r *Recorder[T]) Begin(ctx context.Context, jobID string, status T, apply ...func(next, prev *T)) (bool, error)
- func (r *Recorder[T]) Enabled() bool
- func (r *Recorder[T]) Record(ctx context.Context, jobID string, status T, apply ...func(next, prev *T))
- type RecorderOption
- type Stamper
- type State
- type Status
- type StatusStore
- type Store
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNotFound は、ジョブ状態がまだ記録されていないことを表します。 // 記録前の投入や、この機能より前に作られたジョブでも起こる正常な状態です。 ErrNotFound = errors.New("job status not found") // 表します。 // // ErrNotFound と分けているのは、両者で取るべき判断が正反対だからです。記録が // 無いなら処理を進めてよい一方、読めなかっただけの場合に「無い」とみなすと、 // 完了済みのジョブを未完了と誤認して生成をまるごとやり直します // （Recorder.AlreadySucceeded を参照）。 // // 切り分けは remoteio が返す os.ErrNotExist で行うため、追加のストレージ // 往復は要りません。 ErrUnavailable = errors.New("job status unavailable") // ErrInvalidJobID は、渡されたジョブ ID が正規化を通らなかったことを表します。 // // これが独立していないと、ハンドラーは URL に紛れ込んだ不正な ID をストレージ // 障害と同じ 5xx で返すことになり、再試行しても直らないリクエストを再試行 // させます。原因は jobid のエラーとして残るので、errors.Is で jobid.ErrEmpty / // jobid.ErrTooLong / jobid.ErrInvalidFormat まで辿れます。 ErrInvalidJobID = errors.New("invalid job id") )
Store が返す 3 つの分類です。いずれも原因を包んだまま返すので、errors.Is の 判定はそのままに、ログには元の失敗理由が残ります。
HTTP ハンドラーは次のようにマップしてください。分類ごとに呼び出し側が取るべき 判断が違うため、まとめると必ずどちらかを取り違えます。
ErrInvalidJobID → 400 入力が不正。再試行しても直らない ErrNotFound → 404 未記録。処理を先へ進めてよい ErrUnavailable → 503 読めなかっただけ。あとで読めるかもしれない その他 → 500 壊れた JSON など
Functions ¶
This section is empty.
Types ¶
type Carrier ¶ added in v1.0.2
Carrier は、前回の記録から共通フィールドを引き継ぐために Recorder が使う インターフェースです。
Status を埋め込んだ型は、メソッドの昇格によって自動的にこれを満たします。 利用側が明示的に実装する必要はありません。埋め込んでいない型を渡した場合、 引き継ぎは行われず、渡された値がそのまま保存されます。
type Locator ¶
Locator は、正規化済みのジョブ ID から状態ファイルの URI を組み立てます。
渡されるジョブ ID は Store 側で jobid.Sanitize を通した後の値です。 Locator の中で改めて検証する必要はありません。
func UnderJobDir ¶
UnderJobDir は、baseURI/{jobID}/status.json を指す Locator を返します。
成果物と同じジョブディレクトリ配下に置くため、履歴削除（プレフィックスの一括削除）で 状態ファイルも自動的に片付きます。
type Recorder ¶ added in v1.0.2
type Recorder[T any] struct { // contains filtered or unexported fields }
Recorder は、ワーカーがジョブの進行に合わせて状態を記録するための薄い層です。
移植元の 3 サービスは、いずれも Store の上に同じ 3 つの振る舞いを重ねていました。 個別に持つ必然性がないため、ここへ集約しています。
- 完了済みジョブの再実行ガード（AlreadySucceeded）
- 前回記録からの共通フィールドの引き継ぎ（Record）
- 記録の失敗で生成そのものを止めない（警告ログに留める）
3 は、状態はあくまで観測のための記録であり、書けなかったことを理由に生成を 中断するほうが害が大きいためです。
ワーカーの入口では 1 と 2 が必ず並んで呼ばれるため、その組み合わせは Begin に まとめてあります。
func NewRecorder ¶ added in v1.0.2
func NewRecorder[T any](store StatusStore[T], opts ...RecorderOption) *Recorder[T]
NewRecorder は Recorder を構築します。
store が nil の場合、記録は行われません（Enabled が false を返し、他のメソッドは 何もしません）。状態の記録を任意機能として組み込めるようにするためのもので、 利用側が呼び出しのたびに nil を確かめる必要はありません。
func (*Recorder[T]) AlreadySucceeded ¶ added in v1.0.2
AlreadySucceeded は、そのジョブが既に完了しているかどうかを返します。
Cloud Tasks は at-least-once 配信なので、通知の失敗などでワーカーがエラーを返すと 同じタスクが再配信されます。生成をまるごと呼び直すと生成コストがそのまま二重に 発生するため、完了済みならワーカー側でここで打ち切ってください。
done, err := rec.AlreadySucceeded(ctx, task.JobID)
if err != nil {
return err // 判定できないので再配信に委ねる
}
if done {
return nil
}
未記録（ErrNotFound）は「完了していない」として false を返します。記録が無いのは 記録前の投入やこの機能より前のジョブでも起こる、正常な状態だからです。
状態を読めなかった場合（ErrUnavailable）はエラーを返します。ここで false に倒すと 完了済みのジョブを未完了と誤認して生成をやり直し、このガードが防ぐはずのコストを ガード自身が発生させます。逆に true に倒すと、未完了のジョブを完了扱いにして タスクを ACK してしまい、そのジョブは二度と実行されません。どちらへ倒しても 誤りうるので判断を返し、呼び出し側がエラーを返して再配信に委ねられるようにします。
これはジョブ単位のガードです。処理を継続タスクへ分割して実行する場合、その途中 （state=running）での再配信までは防げません。
func (*Recorder[T]) Begin ¶ added in v1.2.0
func (r *Recorder[T]) Begin(ctx context.Context, jobID string, status T, apply ...func(next, prev *T)) (bool, error)
Begin は、再実行ガードと処理開始の記録を 1 回の読み取りで行います。 完了済みで何もしなかった場合に true を返します。
AlreadySucceeded で打ち切りを判定してから Record で running を書くと、Record が 引き継ぎのために前回の記録を読み直すため、同じ status.json を 1 タスクで 2 度 取得することになります。判定に要る情報は引き継ぎ元とまったく同じなので、 ここでまとめます。記録を伴わない打ち切り判定だけが要るときは AlreadySucceeded を 使ってください。
done, err := rec.Begin(ctx, task.JobID, newStatus(task, StateRunning),
func(next, _ *JobStatus) { next.Attempts++ })
if err != nil {
return err // 判定できないので再配信に委ねる
}
if done {
return nil
}
打ち切りの判断は AlreadySucceeded と同じです。未記録は「完了していない」として false を返し、読めなかった場合はエラーを返します（理由は AlreadySucceeded を 参照）。エラーを返すときは記録も行いません。引き継ぎ元を失ったまま Attempts と QueuedAt をリセットした running を書き残しても、呼び出し側はそのまま再配信へ 委ねるだけだからです。
渡す status は処理開始（running）を表すものにしてください。投入（queued）の記録は 再実行ガードの対象ではないので Record を使います。
func (*Recorder[T]) Record ¶ added in v1.0.2
func (r *Recorder[T]) Record(ctx context.Context, jobID string, status T, apply ...func(next, prev *T))
Record は、前回の記録から共通フィールドを引き継いだうえで status を保存します。
引き継ぐのは Attempts・QueuedAt と、status 側が空のときの Title です （CarryOver を参照）。ワーカーは毎回タスクから状態を組み立て直すため、これが 無いと再試行のたびに試行回数と投入時刻が失われます。
apply は引き継ぎの後、保存の前に呼ばれます。試行回数の加算や、サービス固有 フィールドの引き継ぎに使ってください。prev は前回の記録で、読めなかったときは nil です。未記録（ErrNotFound）は初回の記録として黙って進み、それ以外の読み取り 失敗は引き継ぎ元を失っている（Attempts・QueuedAt がこの記録でリセットされる）ため、 警告ログを残します。
// 投入を記録し、前回の成果物の在り処を残す
rec.Record(ctx, req.JobID, newStatus(req, StateQueued), func(next, prev *JobStatus) {
if prev != nil {
next.OutputDir = prev.OutputDir
}
})
前回が終了済み（IsTerminal）のとき、今回が running・failed なら保存しません （queued は例外です。理由は rolledBack を参照）。判定に使う前回の記録は引き継ぎの ために既に読んでいるので、ストレージへの往復は増えません。
処理開始（running）の記録は Begin を使ってください。再実行ガードと同じ読み取りに 相乗りできます。
保存に失敗しても呼び出し側へは伝えず、警告ログに留めます。
type RecorderOption ¶ added in v1.0.2
type RecorderOption func(*recorderOptions)
RecorderOption は Recorder の挙動を変更します。
func WithLogger ¶ added in v1.0.2
func WithLogger(logger *slog.Logger) RecorderOption
WithLogger は、記録の失敗を書き出すロガーを差し替えます。 既定は slog.Default() です。
type Stamper ¶
Stamper は、Store が共通フィールドを維持するために使うインターフェースです。
Status を埋め込んだ型は、ポインタメソッドの昇格によって自動的にこれを満たします。 利用側が明示的に実装する必要はありません。 逆に Status を埋め込んでいない型を Store に渡した場合、打刻は行われず 渡された値がそのまま保存されます。
type State ¶
type State string
State はジョブのライフサイクル上の状態です。
const ( // StateQueued はキューへ投入済みで、まだワーカーが処理を始めていない状態です。 StateQueued State = "queued" // StateRunning はワーカーが処理中の状態です。 // 継続タスクへ分割して実行される処理では、引き継がれている間もこの状態です。 StateRunning State = "running" // StateSucceeded は成果物の保存まで完了した状態です。 StateSucceeded State = "succeeded" // StateFailed は処理が失敗した状態です。Cloud Tasks による再試行の対象になり得ます。 StateFailed State = "failed" )
type Status ¶
type Status struct {
JobID string `json:"job_id"`
Command string `json:"command,omitempty"`
State State `json:"state"`
// Title は生成対象の題目が確定した時点で埋まります。
Title string `json:"title,omitempty"`
// Error は State が failed のときの失敗理由です。
Error string `json:"error,omitempty"`
// Attempts はワーカーが処理を開始した回数です。2 以上なら再試行されています。
Attempts int `json:"attempts,omitempty"`
QueuedAt time.Time `json:"queued_at,omitzero"`
UpdatedAt time.Time `json:"updated_at"`
}
Status は、アプリ間で共通するジョブ進行状況のフィールドです。
成果物の保存先はサービスごとに形が違う（単一 URI・出力ディレクトリ・複数 URI）ため、 ここには持たせません。利用側は本構造体を埋め込んだ型を定義してください。 Go は埋め込み構造体を JSON でフラットに展開するため、既存の status.json を そのまま読み書きできます。
type JobStatus struct {
jobstatus.Status
OutputDir string `json:"output_dir,omitempty"`
}
func (*Status) CarryOver ¶ added in v1.0.2
CarryOver は、前回の記録から引き継ぐべき共通フィールドを取り込みます。
ワーカーは状態が変わるたびにタスクから状態を組み立て直すため、これが無いと 再試行のたびに試行回数と投入時刻が失われます。Title は、今回の組み立てで 埋まっていない場合にだけ引き継ぎます（生成の途中で題目が確定するサービスが あるため、新しく判明した題目を古い値で上書きしないようにするためです）。
State・Error・UpdatedAt は引き継ぎません。いずれも「今回の記録」を表す値で、 引き継ぐと成功後に古い失敗理由が残り続けます。
func (Status) Common ¶ added in v1.0.2
Common は、埋め込まれた共通フィールドをそのまま返します。
Status を埋め込んだサービス固有の型から、共通部分だけを型引数なしで取り出すための ものです（埋め込みによりメソッドが昇格するため、利用側の実装は要りません）。
func (*Status) EnsureJobID ¶
EnsureJobID は、JobID が空のときだけ補います。 job_id を持たない古い記録を読んだときに、呼び出し側が ID 無しの構造体を 受け取らないようにするためのものです。
func (Status) IsTerminal ¶
IsTerminal は、これ以上状態が変化しない（ポーリングを止めてよい）かどうかを返します。
failed は Cloud Tasks が再試行しうるため終了とはみなしません。
type StatusStore ¶ added in v1.0.2
type StatusStore[T any] interface { Get(ctx context.Context, jobID string) (T, error) Save(ctx context.Context, jobID string, status T) error }
StatusStore は、Recorder が必要とする最小限の読み書きです。 *Store[T] がそのまま満たします。
Recorder が Store 型ではなくインターフェースを受けるのは、利用側が状態の保存先を 差し替えられるようにするためです（テストの偽実装や、ストレージ以外への記録）。
type Store ¶
type Store[T any] struct { // contains filtered or unexported fields }
Store は、リモートストレージを裏付けとしたジョブ状態の読み書きを行います。
型引数 T にはアプリ固有の状態型（Status を埋め込んだ構造体）を指定します。
func NewStore ¶
func NewStore[T any](reader remoteio.Reader, writer remoteio.OutputWriter, locate Locator) *Store[T]
NewStore は Store を構築します。
reader / writer は remoteio.InputReader / remoteio.OutputWriter をそのまま渡せます。
型引数にポインタ型を渡した場合は panic します（mustNotBePointer を参照）。
func (*Store[T]) Delete ¶
Delete はジョブ状態を削除します。履歴削除に追随させるために使います。
writer が未設定のときは何もしません（記録できていない以上、消すものも無いため）。
func (*Store[T]) Get ¶
Get はジョブ状態を取得します。未記録の場合は ErrNotFound を返します。
壊れた JSON は未記録ではなくデコード失敗として返します。未記録と同じ扱いにすると、 破損に気づかないまま再生成が走り続けるためです。
デコードは encoding/json/v2 の UnmarshalRead で行います。JSON 値のあとに残った バイト列と、重複したキーを拒否するためです。従来の Decoder はどちらも黙って 受け取っていました。とくに重複キーは後勝ちで、途中で切れた書き込みに次の書き込みが 続いた status.json が {"state":"succeeded",...,"state":"running"} の形になると、 running として読めてしまいます。これは再実行ガードが防いでいるはずの巻き戻しが、 記録ではなく読み取りの側から入ってくる経路です。
Source Files