一覧PG-20-2 · データ処理の組立て

再試行と冪等性

固定の成功・失敗列で試行回数と待機予定を確認し、その後、応答を失った加算を再送します。同じ要求を識別するキーを保持して、更新を一度だけ反映する方法と、同じキーで内容が違う場合の拒否を確認します。

説明

応答が返らないときでも、相手側の更新が失敗したとは限りません。処理は成功して応答だけを失った場合、単純な再送で同じ更新を二度行う可能性があります。

1失敗の種類を表す

一時的な失敗を区別するための例外型を定義します。

class TemporaryError(Exception):再試行対象の失敗型


    pass

再試行の対象をこの例外へ限定します。入力不正など、同じ要求を繰り返しても直らない失敗まで一律に再試行しない設計です。

2試行上限と待機予定を定義する

attemptsは初回を含む最大試行数です。待機は実時間ではなく、次の試行までの予定秒数として一覧へ記録します。

def retry(operation, attempts, waits):
    for attempt in

range(1, attempts + 1)初回を含む試行番号

:
        try:
            print("attempt", attempt)

return operation()成功したら終了

except TemporaryError対象の失敗だけを再試行

if attempt == attempts:最後の失敗は外へ伝える


                raise

waits.append(min(2 ** (attempt - 1), 4))指数増加と4秒の上限

初回の後は1秒、次は2秒、以後は上限4秒です。最後の失敗後は次の試行がないため、待機を追加せず例外を伝えます。この関数はattemptsが1以上という呼出し条件で使います。

3決めた結果列で操作を作る

一時失敗、入力不正、成功値を順に返す関数を作ります。

def scripted(outcomes):

remaining = iter(outcomes)この処理専用の結果列


    def operation():

outcome = next(remaining)一回分の結果を取得


        if outcome == "temporary":

raise TemporaryError一時失敗

("try later")
        if outcome == "invalid":

raise ValueError入力不正

("invalid request")
        return outcome

return operation結果列を保持した処理を返す

この固定列により、ネットワークの偶然の遅延に頼らず、何回目で成功したかを確認できます。関数を生成するたびに独立した結果列を持ちます。

4三回目で成功する要求

一時失敗が二回続いた後でokになる処理を、最大4回で実行します。

waits = []; result = retry(scripted(["temporary", "temporary", "ok"]), 4, waits)最大4回・三回目に成功; print(result, waits, sum(waits)待機予定と合計)

試行は3回、再試行は2回です。待機予定は1秒と2秒、合計3秒になります。成功後に余分な試行や待機は追加されません。

5上限まで失敗する要求

最大5回を全て一時失敗させます。

try:
    waits.clear()

retry(scripted(["temporary"] * 5), 5, waits)最大5回を全て失敗させる


except TemporaryError:
    print("failed",

waits, sum(waits)上限付きの待機予定)

5回試行して失敗を返し、待機は1、2、4、4の4回分で合計11秒です。指数バックオフだけでは同時に失敗した多数の処理の再試行時点が揃うため、実際の設計では待機へ揺らぎを加える方法も使われます。

6再試行しない失敗

最初の結果を入力不正にし、その後にokを用意します。

try:
    waits.clear()

retry(scripted(["invalid", "ok"]), 4, waits)最初に入力不正

except ValueError as error再試行対象外の失敗

:
    print(str(error),

waits待機は追加されない)

試行1でValueErrorが外へ伝わり、待機は空です。同じ不正な要求を繰り返しても解決しない失敗を、回数上限まで続けていません。

7更新後に応答を失う操作

加算は済んでいるのに、応答を失ったとして一時失敗を返す関数です。

def add_without_key(state, amount, lose_reply=False):

state["total"] += amount先に加算を反映

if lose_reply:応答だけを失う条件

raise TemporaryError("reply lost")呼出し側には失敗が届く


    return state["total"]

呼出し側が受け取る例外だけを見ると、更新前に失敗したか、更新後に応答だけを失ったかを判断できません。

8同じ加算を再送する

初回は応答を失わせ、同じ10を再度加算します。

try:
    state = {"total": 0}

add_without_key(state, 10, lose_reply=True)加算後に応答を失う


except TemporaryError:

print("after timeout", state)相手側に残る更新


    print("retry result",

add_without_key(state, 10)同じ意図を再送)

タイムアウト時点ですでに10、再送後は20です。要求を識別せず加算を繰り返すと、同じ一回の意図が二回分として反映されます。

9要求キーと内容・結果を保存する

キーごとに金額と返却結果を保存します。同じキーで金額が違う場合は拒否します。

def add_once(state, key, amount, lose_reply=False):

if key in state["seen"]:処理済みの要求を探す


        saved_amount, result = state["seen"][key]

if saved_amount != amount:同じキーの内容違いを拒否


            raise ValueError("key reused with different amount")

return result再送には保存済みの結果


    state["total"] += amount
    result = state["total"]

state["seen"][key] = (amount, result)初回の内容と結果を記録


    if lose_reply:
        raise TemporaryError("reply lost")
    return result

初回に更新と要求の記録を行い、再送時は保存済みの結果を返します。この一つの処理内の例を外部ストレージへ移す場合には、更新と記録の原子性、同時要求、キーの有効範囲・保存期限も必要です。

10応答を失っても加算は一回

req-1で10を加算し、応答を失った後に同じキーで再送します。

try:
    protected = {"total": 0, "seen": {}}

add_once(protected, "req-1", 10, lose_reply=True)キー付きの更新後に応答を失う


except TemporaryError:
    print("after timeout", protected)
    print("retry result",

add_once(protected, "req-1", 10)同じキーで再送

)
    print("total",

protected["total"]一回分だけ反映した合計)

合計は10のままです。再送は保存済みの結果10を返し、二回目の加算を行いません。応答を受け取れなかった場合でも、同じキーで同じ要求を再送することで重複を避けています。

11キーの再利用と別の要求

req-1の金額を20へ変える要求は拒否し、新しいreq-2の10は別の更新として扱います。

try:

add_once(protected, "req-1", 20)同じキーで異なる内容


except ValueError as error:
    print(str(error),

protected["total"]拒否後の合計

)
    print("new request",

add_once(protected, "req-2", 10)新しいキーの別要求)

内容違いは拒否されて合計10を保ち、新しいキーの要求で20になります。同じ金額であることだけでは、同じ要求かどうかは決められません。

まとめ

初回を含む試行上限と、再試行の対象となる失敗を明示します。待機は最後の失敗後へ余分に追加しません。

応答を失っても更新が成功している場合があります。加算の単純な再送は重複更新を起こします。

同じ要求のキー・内容・結果を保持し、再送には記録済みの結果を返します。永続化や同時実行では記録と更新を一体で扱う設計が必要です。

操作履歴と解説を振り返れます。

Python 3

class TemporaryError(Exception):
    pass

def retry(operation, attempts, waits):
    for attempt in range(1, attempts + 1):
        try:
            print("attempt", attempt)
            return operation()
        except TemporaryError:
            if attempt == attempts:
                raise
            waits.append(min(2 ** (attempt - 1), 4))

def scripted(outcomes):
    remaining = iter(outcomes)
    def operation():
        outcome = next(remaining)
        if outcome == "temporary":
            raise TemporaryError("try later")
        if outcome == "invalid":
            raise ValueError("invalid request")
        return outcome
    return operation

waits = []; result = retry(scripted(["temporary", "temporary", "ok"]), 4, waits); print(result, waits, sum(waits))

attempt 1
attempt 2
attempt 3
ok [1, 2] 3

try:
    waits.clear()
    retry(scripted(["temporary"] * 5), 5, waits)
except TemporaryError:
    print("failed", waits, sum(waits))

attempt 1
attempt 2
attempt 3
attempt 4
attempt 5
failed [1, 2, 4, 4] 11

try:
    waits.clear()
    retry(scripted(["invalid", "ok"]), 4, waits)
except ValueError as error:
    print(str(error), waits)

attempt 1
invalid request []

def add_without_key(state, amount, lose_reply=False):
    state["total"] += amount
    if lose_reply:
        raise TemporaryError("reply lost")
    return state["total"]

try:
    state = {"total": 0}
    add_without_key(state, 10, lose_reply=True)
except TemporaryError:
    print("after timeout", state)
    print("retry result", add_without_key(state, 10))

after timeout {'total': 10}
retry result 20

def add_once(state, key, amount, lose_reply=False):
    if key in state["seen"]:
        saved_amount, result = state["seen"][key]
        if saved_amount != amount:
            raise ValueError("key reused with different amount")
        return result
    state["total"] += amount
    result = state["total"]
    state["seen"][key] = (amount, result)
    if lose_reply:
        raise TemporaryError("reply lost")
    return result

try:
    protected = {"total": 0, "seen": {}}
    add_once(protected, "req-1", 10, lose_reply=True)
except TemporaryError:
    print("after timeout", protected)
    print("retry result", add_once(protected, "req-1", 10))
    print("total", protected["total"])

after timeout {'total': 10, 'seen': {'req-1': (10, 10)}}
retry result 10
total 10

try:
    add_once(protected, "req-1", 20)
except ValueError as error:
    print(str(error), protected["total"])
    print("new request", add_once(protected, "req-2", 10))

key reused with different amount 10
new request 20

演習完了