隊列模型 ModelWithQueue 源碼解析：status、retry_at 與原子租約機制)
ArchiveBox 耐用任務(wù)隊列模型 ModelWithQueue 源碼解析status、retry_at 與原子租約機制【免費下載鏈接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...項目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox本文以 ArchiveBox 的archivebox.workers.models模塊API 文檔見 archivebox.workers.models.md實現(xiàn)見 workers/models.py為核心深入講解 ArchiveBox 調(diào)度系統(tǒng)賴以運轉(zhuǎn)的耐用隊列durable queue協(xié)議status/retry_at雙字段狀態(tài)模型、pause/resume生命周期控制、基于條件更新的原子租約lease與安全更新機制以及它們在 Crawl、Snapshot、Binary 等具體模型和 runner 運行器中的實際調(diào)用方式。讀完本文你將掌握 ArchiveBox 分布式/多進程調(diào)度中誰擁有任務(wù)、任務(wù)何時重試、如何防止重復(fù)執(zhí)行的底層實現(xiàn)原理并了解如何在自己的 Django 模型上復(fù)用這套隊列協(xié)議。模塊定位調(diào)度系統(tǒng)的公共隊列協(xié)議ArchiveBox 使用 Django ORM 作為任務(wù)調(diào)度的中心化狀態(tài)存儲爬蟲任務(wù)Crawl、頁面快照Snapshot、依賴二進制安裝Binary都圍繞隊列行運轉(zhuǎn)由 runner.py 中的運行器循環(huán)消費。archivebox.workers.models就是這套體系的最小公共部分——它只定義了一個抽象基類ModelWithQueue其模塊 docstring 明確寫道Durable queue fields and atomic lease operations shared by work rows. Concrete models own lifecycle transitions. This mixin only owns the common database queue protocol: status, retry_at, pause/resume, and claims.也就是說具體模型自己決定生命周期狀態(tài)遷移規(guī)則ModelWithQueue只負責提供通用的數(shù)據(jù)庫隊列協(xié)議——狀態(tài)字段、重試時間、暫停/恢復(fù)、以及認領(lǐng)claim操作。這種mixin 管協(xié)議、具體模型管遷移的職責劃分是理解整個 workers 模塊的關(guān)鍵。隊列狀態(tài)協(xié)議DefaultStatusChoices模塊首先定義了默認的狀態(tài)枚舉class DefaultStatusChoices(models.TextChoices): QUEUED queued, Queued STARTED started, Started PAUSED paused, Paused SEALED sealed, Sealed四個狀態(tài)的含義如下狀態(tài)枚舉值含義QUEUEDqueued任務(wù)已入隊等待 worker 認領(lǐng)STARTEDstarted任務(wù)正在處理中活躍狀態(tài)PAUSEDpaused任務(wù)被用戶暫停不再調(diào)度SEALEDsealed任務(wù)已封存終態(tài)不再處理DefaultStatusChoices繼承自 Django 的models.TextChoices因此同時具備枚舉常量DefaultStatusChoices.QUEUED與字段 choices.choices兩套能力。需要說明的是這只是默認協(xié)議——使用方可以通過自定義StatusChoices完全重定義狀態(tài)集見下文 Binary 模型的定制因此這四態(tài)是調(diào)度系統(tǒng)的常規(guī)約定而非鐵律。兩個核心隊列字段與默認值工廠ModelWithQueue只有兩個自有數(shù)據(jù)庫字段它們構(gòu)成了隊列的雙支柱default_status_field: models.CharField models.CharField( choicesDefaultStatusChoices.choices, max_length15, defaultDefaultStatusChoices.QUEUED, nullFalse, blankFalse, db_indexTrue, ) default_retry_at_field: models.DateTimeField models.DateTimeField(defaulttimezone.now, nullTrue, blankTrue, db_indexTrue)statusCharFieldmax_length15任務(wù)的當前狀態(tài)默認QUEUEDnullFalse, blankFalse強制非空且db_indexTrue保證按狀態(tài)過濾的高效性。retry_atDateTimeField可空默認當前時間任務(wù)何時到期可被認領(lǐng)。它同時承擔了到期時間與租約鎖雙重職責下文詳述同樣帶數(shù)據(jù)庫索引。字段本身通過deconstruct()參數(shù)以模塊級默認值工廠的形式定義default_status_field/default_retry_at_field方便各模型復(fù)用。ModelWithQueue內(nèi)部據(jù)此聲明字段status: models.CharField models.CharField(**default_status_field.deconstruct()[3]) retry_at: models.DateTimeField models.DateTimeField(**default_retry_at_field.deconstruct()[3])使用方可以在此基礎(chǔ)上覆蓋參數(shù)例如 Binary 模型調(diào)用ModelWithQueue.StatusField(choicesStatusChoices.choices, defaultStatusChoices.QUEUED, max_length16)見 machine/models.py微調(diào)max_length。兩個關(guān)鍵常量RETRY_AT_MAX datetime(9999, 1, 1, tzinfoUTC) ACTIVE_STATE_LEASE_SECONDS 60RETRY_AT_MAX一個遠未來時間戳語義上等價于永不重試。pause()會把retry_at置為該值從而讓被暫停的任務(wù)在get_queue()查詢中永遠不滿足retry_at__ltenow條件實現(xiàn)不可見即不調(diào)度。ACTIVE_STATE_LEASE_SECONDS 60活躍狀態(tài)的默認租約時長。當 Crawl 保持STARTED狀態(tài)繼續(xù)推進子任務(wù)時runner 會用它刷新父行的租約見 runner.py避免父行租約過期被其他 worker 搶走。模塊還導出了logger模塊日志器、MODULE_PATH/REPO_ROOT/PACKAGE_ROOT用于save()覆蓋中定位調(diào)用方棧幀見下文這些是支撐調(diào)試告警的輔助常量。ModelWithQueue抽象基類的完整 APIModelWithQueue繼承django.db.models.Model其Meta聲明app_label workers且abstract True——它不產(chǎn)生數(shù)據(jù)表只為子類提供字段與方法從倉庫看 workers/migrations 目錄僅有__init__.py正因抽象模型不生成遷移。下面按職責分組介紹其完整 API。類級狀態(tài)常量類屬性默認值說明StatusChoicesDefaultStatusChoices狀態(tài)枚舉子類可覆蓋INITIAL_STATEQUEUED初始狀態(tài)ACTIVE_STATESTARTED活躍狀態(tài)FINAL_STATES(SEALED,)終態(tài)集合FINAL_OR_ACTIVE_STATES(*FINAL_STATES, ACTIVE_STATE)終態(tài) 活躍態(tài)warn_on_save_outside_runnerTrue是否在 runner 進程外save()時告警實例屬性RETRY_AT與STATE只是retry_at、status的屬性別名提供統(tǒng)一的讀寫接口。生命周期操作pause / resume / bump_retry_atdef pause(self, *, save: bool True) - bool: paused_state getattr(self.StatusChoices, PAUSED, None) if paused_state is None or self.status in self.FINAL_STATES or self.is_paused: return False previous_status self.status self.status paused_state self.retry_at RETRY_AT_MAX if save: return self.safe_update( {status: paused_state, retry_at: RETRY_AT_MAX}, extra_filter{status: previous_status}, ) return Truepause()將狀態(tài)置為PAUSED并把retry_at推到RETRY_AT_MAX。三個保護條件——枚舉未定義PAUSED、已是終態(tài)、已處于暫?！紩苯臃祷谾alse。寫庫時使用safe_update且extra_filter{status: previous_status}確保只有仍處于原狀態(tài)的行才被更新并發(fā)安全。resume(whenNone)僅當當前is_paused時才生效把狀態(tài)改回QUEUEDretry_at設(shè)為when默認timezone.now()即立即恢復(fù)調(diào)度同樣帶extra_filter{status: paused_state}的守衛(wèi)。調(diào)用方可以指定when實現(xiàn)定時恢復(fù)。bump_retry_at(seconds10)把retry_at從當前時刻順延 N 秒用于簡單的指數(shù)退避或失敗重試。is_paused判斷當前狀態(tài)是否為枚舉中定義的PAUSED。隊列查詢與原子租約get_queue / claim_for_worker / claim_processing_lock這是模塊最核心的并發(fā)機制全部圍繞retry_at實現(xiàn)誰先到期誰被處理、誰先更新誰擁有classmethod def get_queue(cls): return cls.objects.filter(retry_at__ltetimezone.now()).order_by(retry_at) classmethod def claim_for_worker(cls, obj: ModelWithQueue, lock_seconds: int 60) - bool: now timezone.now() lock_until now timedelta(secondslock_seconds) updated cls.objects.filter(pkobj.pk, retry_atobj.retry_at, retry_at__ltenow).update( retry_atlock_until, modified_atnow, ) if updated 1: obj.retry_at lock_until cast(Any, obj).modified_at now return updated 1get_queue()取出所有到期retry_at now的行并按retry_at升序排列構(gòu)成待處理隊列。這個查詢天然排除了PAUSEDretry_at為遠未來和未來才到期的任務(wù)。claim_for_worker()認領(lǐng)的核心是一個**樂觀鎖 CAS比較并交換**操作。更新條件同時包含pk、retry_atobj.retry_at讀到的舊值和retry_at__ltenow未過期。只有恰好更新 1 行updated 1才算認領(lǐng)成功并把retry_at推進到當前時間 lock_seconds相當于租約到期時間。兩個并發(fā) worker 同時認領(lǐng)同一行時后執(zhí)行的 UPDATE 因retry_at已被前一個改成租約時間而不滿足retry_atobj.retry_at條件更新 0 行即失敗——這就是原子租約防重入的原理。claim_processing_lock(lock_seconds60)實例方法版認領(lǐng)。額外檢查終態(tài)與retry_at is None通過后委托給claim_for_worker。runner 處理每個任務(wù)前都調(diào)用它拿到鎖才繼續(xù)執(zhí)行。安全更新safe_update / update_and_requeuedef safe_update(self, update_fields, *, refreshTrue, extra_filterNone) - bool: values dict(update_fields) values.setdefault(modified_at, timezone.now()) queryset type(self).objects.filter(pkself.pk) if extra_filter: queryset queryset.filter(**extra_filter) updated queryset.update(**values) if updated ! 1 and extra_filter: current type(self).objects.filter(pkself.pk).values(status).first() logger.info( SafeUpdateGuardMiss: %s row %s extra_filter%s current_status%s loaded_status%s ..., ... ) if refresh: try: self.refresh_from_db() except type(self).DoesNotExist: pass return updated 1safe_update()用帶可選extra_filter的 UPDATE 實現(xiàn)帶守衛(wèi)的原子更新返回是否恰好更新 1 行。守衛(wèi)失敗時如extra_filter指定的狀態(tài)已被他人改變會記錄SafeUpdateGuardMiss日志便于排查競態(tài)。默認在更新后refresh_from_db()同步內(nèi)存態(tài)。update_and_requeue(**kwargs)safe_update的特化版本固定extra_filter{retry_at: self.retry_at}——即只有租約未變時才能更新再次確認所有權(quán)后才允許改寫狀態(tài)。runner 中大量使用它推進 Crawl/Snapshot 的調(diào)度狀態(tài)見 runner.py。save() 覆蓋runner 進程外的保存告警ModelWithQueue.save()覆寫了 Django 的save()實現(xiàn)了一個運行期安全網(wǎng)當非新增行在非 ORCHESTRATOR編排器進程中被save()時會通過棧幀回溯利用MODULE_PATH/PACKAGE_ROOT/REPO_ROOT定位調(diào)用者跳過本模塊與 site-packages 幀記錄警告日志指明調(diào)用方文件與行號。其意圖是隊列行的狀態(tài)變更應(yīng)當由 runner 統(tǒng)一管理直接在任意代碼里save()可能繞過租約守衛(wèi)造成狀態(tài)錯亂該告警幫助開發(fā)者盡早發(fā)現(xiàn)這類越權(quán)寫庫。子類可通過warn_on_save_outside_runner False關(guān)閉Binary 模型即如此因為它由BinaryService同步驅(qū)動安裝流程見 machine/models.py。輔助方法status_counts / extend_choices / StatusField / RetryAtFieldstatus_counts(querysetNone, statusesNone) - dict[str, int]類方法按狀態(tài)統(tǒng)計行數(shù)返回如{queued: 12, started: 3, ...}用于調(diào)度概覽與 admin 面板展示。extend_choices(base_choices)一個類裝飾器工廠把基礎(chǔ)枚舉與附加枚舉合并生成新的TextChoices供需要擴展狀態(tài)集的模型使用。StatusField(**kwargs)/RetryAtField(**kwargs)類方法字段工廠以默認字段定義為基礎(chǔ)合并覆蓋參數(shù)子類聲明字段時直接復(fù)用Crawl、Snapshot、Binary 三處均如此。子類化實踐Crawl、Snapshot 與 Binary從倉庫看ModelWithQueue被三個模型繼承展示了默認協(xié)議復(fù)用 按需定制的完整模式Crawl 與 Snapshot直接復(fù)用默認四態(tài)Crawl 與 Snapshot 都按默認協(xié)議定制# crawls/models.py class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): status ModelWithQueue.StatusField( choicesModelWithQueue.StatusChoices, defaultModelWithQueue.StatusChoices.QUEUED, ) retry_at ModelWithQueue.RetryAtField(defaulttimezone.now) StatusChoices ModelWithQueue.StatusChoices INITIAL_STATE StatusChoices.QUEUED ACTIVE_STATE StatusChoices.STARTED FINAL_STATES (StatusChoices.SEALED,) FINAL_OR_ACTIVE_STATES (*FINAL_STATES, ACTIVE_STATE)Snapshot 結(jié)構(gòu)相同core/models.py并額外定義RUNNABLE_STATES (QUEUED, STARTED)、OPEN_STATES等派生集合。兩者共享默認四態(tài)queued → started → sealed可被paused中斷。Binary完全重定義狀態(tài)機Binary 演示了最激進的定制——它只用兩態(tài)class StatusChoices(models.TextChoices): QUEUED queued, Queued INSTALLED installed, Installed status ModelWithQueue.StatusField(choicesStatusChoices.choices, defaultStatusChoices.QUEUED, max_length16) INITIAL_STATE StatusChoices.QUEUED ACTIVE_STATE StatusChoices.QUEUED # 活躍態(tài)即 QUEUED FINAL_STATES (StatusChoices.INSTALLED,) warn_on_save_outside_runner FalseBinary 的模塊 docstring 說明了設(shè)計意圖安裝流程是同步的queued → installed轉(zhuǎn)換失敗時保持queued并把retry_at順延以便稍后重試數(shù)據(jù)庫行是唯一的生命周期狀態(tài)worker 中斷后無需在內(nèi)存中二次協(xié)調(diào)狀態(tài)機。它還演示了update_and_requeue(retry_atNone, statusINSTALLED)的收尾用法machine/models.py。運行器中的真實調(diào)用鏈runnerservices/runner.py是把這套協(xié)議串起來的消費端其處理每個到期任務(wù)的標準流程是選行通過retry_at__ltenow類查詢等價于get_queue取出到期任務(wù)。認領(lǐng)調(diào)用crawl.claim_processing_lock(lock_secondslock_seconds)L1361、L1395或Snapshot.claim_for_worker(snapshot, lock_secondslock_seconds)L1448原子獲取租約失敗即返回False說明該行已被其他 worker 認領(lǐng)。執(zhí)行與續(xù)租處理期間用crawl.update_and_requeue(statusSTARTED, retry_atnow timedelta(secondsACTIVE_STATE_LEASE_SECONDS))續(xù)租L1352把租約從 60 秒起步按需延長。收尾終態(tài)任務(wù)如SEALED的 Crawl在claim_for_worker成功后執(zhí)行cleanup_runtime()最后update_and_requeue(retry_atNone)清空調(diào)度時間使其退出隊列L1405-L1411。值得注意的邊界處理當 Snapshot 的租約已過期但關(guān)聯(lián)的abx-dl鉤子進程仍在運行process_set.filter(statusrunning)時runner 不會啟動第二波處理而是update_and_requeue(retry_atnow lock_seconds)重新排隊L1471-L1479保留快照級的所有權(quán)邊界防止重復(fù)執(zhí)行。測試驗證與可觀測性倉庫測試對這套協(xié)議有直接覆蓋可作為理解佐證test_crawl_runner.py 斷言snapshot.claim_processing_lock(lock_seconds60) is True驗證單 worker 場景下的認領(lǐng)成功路徑同文件 L414 使用crawl.update_and_requeue(statusQUEUED, retry_attimezone.now())構(gòu)造調(diào)度場景驗證重排隊邏輯。此外認領(lǐng)失敗與守衛(wèi)未命中都會留下可觀測日志SafeUpdateGuardMisssafe_update 守衛(wèi)失敗以及save() outside runner process告警配合status_counts()的狀態(tài)分布統(tǒng)計可以在運維層面快速定位任務(wù)堆積在哪一態(tài)、是否被越權(quán)改寫。小結(jié)archivebox.workers.models用約 230 行代碼給出了一個干凈、可復(fù)用的數(shù)據(jù)庫級任務(wù)隊列協(xié)議以statusretry_at兩個索引字段為存儲以條件 UPDATE 影響行數(shù)判定實現(xiàn)原子認領(lǐng)與安全更新以pause/resume與RETRY_AT_MAX實現(xiàn)暫停語義以ACTIVE_STATE_LEASE_SECONDS維持活躍任務(wù)租約。Crawl、Snapshot、Binary 通過繼承與覆蓋各取所需runner 則作為唯一合法的狀態(tài)遷移入口消費隊列——這套mixin 管協(xié)議、模型管遷移、runner 管執(zhí)行的設(shè)計是 ArchiveBox 多進程調(diào)度一致性的基石。【免費下載鏈接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...項目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 ArchiveBox 耐用任務(wù)隊列模型 ModelWithQueue 源碼解析：status、retry_at 與原子租約機制
ArchiveBox 耐用任務(wù)隊列模型 ModelWithQueue 源碼解析status、retry_at 與原子租約機制【免費下載鏈接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...項目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox本文以 ArchiveBox 的archivebox.workers.models模塊API 文檔見 archivebox.workers.models.md實現(xiàn)見 workers/models.py為核心深入講解 ArchiveBox 調(diào)度系統(tǒng)賴以運轉(zhuǎn)的耐用隊列durable queue協(xié)議status/retry_at雙字段狀態(tài)模型、pause/resume生命周期控制、基于條件更新的原子租約lease與安全更新機制以及它們在 Crawl、Snapshot、Binary 等具體模型和 runner 運行器中的實際調(diào)用方式。讀完本文你將掌握 ArchiveBox 分布式/多進程調(diào)度中誰擁有任務(wù)、任務(wù)何時重試、如何防止重復(fù)執(zhí)行的底層實現(xiàn)原理并了解如何在自己的 Django 模型上復(fù)用這套隊列協(xié)議。模塊定位調(diào)度系統(tǒng)的公共隊列協(xié)議ArchiveBox 使用 Django ORM 作為任務(wù)調(diào)度的中心化狀態(tài)存儲爬蟲任務(wù)Crawl、頁面快照Snapshot、依賴二進制安裝Binary都圍繞隊列行運轉(zhuǎn)由 runner.py 中的運行器循環(huán)消費。archivebox.workers.models就是這套體系的最小公共部分——它只定義了一個抽象基類ModelWithQueue其模塊 docstring 明確寫道Durable queue fields and atomic lease operations shared by work rows. Concrete models own lifecycle transitions. This mixin only owns the common database queue protocol: status, retry_at, pause/resume, and claims.也就是說具體模型自己決定生命周期狀態(tài)遷移規(guī)則ModelWithQueue只負責提供通用的數(shù)據(jù)庫隊列協(xié)議——狀態(tài)字段、重試時間、暫停/恢復(fù)、以及認領(lǐng)claim操作。這種mixin 管協(xié)議、具體模型管遷移的職責劃分是理解整個 workers 模塊的關(guān)鍵。隊列狀態(tài)協(xié)議DefaultStatusChoices模塊首先定義了默認的狀態(tài)枚舉class DefaultStatusChoices(models.TextChoices): QUEUED queued, Queued STARTED started, Started PAUSED paused, Paused SEALED sealed, Sealed四個狀態(tài)的含義如下狀態(tài)枚舉值含義QUEUEDqueued任務(wù)已入隊等待 worker 認領(lǐng)STARTEDstarted任務(wù)正在處理中活躍狀態(tài)PAUSEDpaused任務(wù)被用戶暫停不再調(diào)度SEALEDsealed任務(wù)已封存終態(tài)不再處理DefaultStatusChoices繼承自 Django 的models.TextChoices因此同時具備枚舉常量DefaultStatusChoices.QUEUED與字段 choices.choices兩套能力。需要說明的是這只是默認協(xié)議——使用方可以通過自定義StatusChoices完全重定義狀態(tài)集見下文 Binary 模型的定制因此這四態(tài)是調(diào)度系統(tǒng)的常規(guī)約定而非鐵律。兩個核心隊列字段與默認值工廠ModelWithQueue只有兩個自有數(shù)據(jù)庫字段它們構(gòu)成了隊列的雙支柱default_status_field: models.CharField models.CharField( choicesDefaultStatusChoices.choices, max_length15, defaultDefaultStatusChoices.QUEUED, nullFalse, blankFalse, db_indexTrue, ) default_retry_at_field: models.DateTimeField models.DateTimeField(defaulttimezone.now, nullTrue, blankTrue, db_indexTrue)statusCharFieldmax_length15任務(wù)的當前狀態(tài)默認QUEUEDnullFalse, blankFalse強制非空且db_indexTrue保證按狀態(tài)過濾的高效性。retry_atDateTimeField可空默認當前時間任務(wù)何時到期可被認領(lǐng)。它同時承擔了到期時間與租約鎖雙重職責下文詳述同樣帶數(shù)據(jù)庫索引。字段本身通過deconstruct()參數(shù)以模塊級默認值工廠的形式定義default_status_field/default_retry_at_field方便各模型復(fù)用。ModelWithQueue內(nèi)部據(jù)此聲明字段status: models.CharField models.CharField(**default_status_field.deconstruct()[3]) retry_at: models.DateTimeField models.DateTimeField(**default_retry_at_field.deconstruct()[3])使用方可以在此基礎(chǔ)上覆蓋參數(shù)例如 Binary 模型調(diào)用ModelWithQueue.StatusField(choicesStatusChoices.choices, defaultStatusChoices.QUEUED, max_length16)見 machine/models.py微調(diào)max_length。兩個關(guān)鍵常量RETRY_AT_MAX datetime(9999, 1, 1, tzinfoUTC) ACTIVE_STATE_LEASE_SECONDS 60RETRY_AT_MAX一個遠未來時間戳語義上等價于永不重試。pause()會把retry_at置為該值從而讓被暫停的任務(wù)在get_queue()查詢中永遠不滿足retry_at__ltenow條件實現(xiàn)不可見即不調(diào)度。ACTIVE_STATE_LEASE_SECONDS 60活躍狀態(tài)的默認租約時長。當 Crawl 保持STARTED狀態(tài)繼續(xù)推進子任務(wù)時runner 會用它刷新父行的租約見 runner.py避免父行租約過期被其他 worker 搶走。模塊還導出了logger模塊日志器、MODULE_PATH/REPO_ROOT/PACKAGE_ROOT用于save()覆蓋中定位調(diào)用方棧幀見下文這些是支撐調(diào)試告警的輔助常量。ModelWithQueue抽象基類的完整 APIModelWithQueue繼承django.db.models.Model其Meta聲明app_label workers且abstract True——它不產(chǎn)生數(shù)據(jù)表只為子類提供字段與方法從倉庫看 workers/migrations 目錄僅有__init__.py正因抽象模型不生成遷移。下面按職責分組介紹其完整 API。類級狀態(tài)常量類屬性默認值說明StatusChoicesDefaultStatusChoices狀態(tài)枚舉子類可覆蓋INITIAL_STATEQUEUED初始狀態(tài)ACTIVE_STATESTARTED活躍狀態(tài)FINAL_STATES(SEALED,)終態(tài)集合FINAL_OR_ACTIVE_STATES(*FINAL_STATES, ACTIVE_STATE)終態(tài) 活躍態(tài)warn_on_save_outside_runnerTrue是否在 runner 進程外save()時告警實例屬性RETRY_AT與STATE只是retry_at、status的屬性別名提供統(tǒng)一的讀寫接口。生命周期操作pause / resume / bump_retry_atdef pause(self, *, save: bool True) - bool: paused_state getattr(self.StatusChoices, PAUSED, None) if paused_state is None or self.status in self.FINAL_STATES or self.is_paused: return False previous_status self.status self.status paused_state self.retry_at RETRY_AT_MAX if save: return self.safe_update( {status: paused_state, retry_at: RETRY_AT_MAX}, extra_filter{status: previous_status}, ) return Truepause()將狀態(tài)置為PAUSED并把retry_at推到RETRY_AT_MAX。三個保護條件——枚舉未定義PAUSED、已是終態(tài)、已處于暫?！紩苯臃祷谾alse。寫庫時使用safe_update且extra_filter{status: previous_status}確保只有仍處于原狀態(tài)的行才被更新并發(fā)安全。resume(whenNone)僅當當前is_paused時才生效把狀態(tài)改回QUEUEDretry_at設(shè)為when默認timezone.now()即立即恢復(fù)調(diào)度同樣帶extra_filter{status: paused_state}的守衛(wèi)。調(diào)用方可以指定when實現(xiàn)定時恢復(fù)。bump_retry_at(seconds10)把retry_at從當前時刻順延 N 秒用于簡單的指數(shù)退避或失敗重試。is_paused判斷當前狀態(tài)是否為枚舉中定義的PAUSED。隊列查詢與原子租約get_queue / claim_for_worker / claim_processing_lock這是模塊最核心的并發(fā)機制全部圍繞retry_at實現(xiàn)誰先到期誰被處理、誰先更新誰擁有classmethod def get_queue(cls): return cls.objects.filter(retry_at__ltetimezone.now()).order_by(retry_at) classmethod def claim_for_worker(cls, obj: ModelWithQueue, lock_seconds: int 60) - bool: now timezone.now() lock_until now timedelta(secondslock_seconds) updated cls.objects.filter(pkobj.pk, retry_atobj.retry_at, retry_at__ltenow).update( retry_atlock_until, modified_atnow, ) if updated 1: obj.retry_at lock_until cast(Any, obj).modified_at now return updated 1get_queue()取出所有到期retry_at now的行并按retry_at升序排列構(gòu)成待處理隊列。這個查詢天然排除了PAUSEDretry_at為遠未來和未來才到期的任務(wù)。claim_for_worker()認領(lǐng)的核心是一個**樂觀鎖 CAS比較并交換**操作。更新條件同時包含pk、retry_atobj.retry_at讀到的舊值和retry_at__ltenow未過期。只有恰好更新 1 行updated 1才算認領(lǐng)成功并把retry_at推進到當前時間 lock_seconds相當于租約到期時間。兩個并發(fā) worker 同時認領(lǐng)同一行時后執(zhí)行的 UPDATE 因retry_at已被前一個改成租約時間而不滿足retry_atobj.retry_at條件更新 0 行即失敗——這就是原子租約防重入的原理。claim_processing_lock(lock_seconds60)實例方法版認領(lǐng)。額外檢查終態(tài)與retry_at is None通過后委托給claim_for_worker。runner 處理每個任務(wù)前都調(diào)用它拿到鎖才繼續(xù)執(zhí)行。安全更新safe_update / update_and_requeuedef safe_update(self, update_fields, *, refreshTrue, extra_filterNone) - bool: values dict(update_fields) values.setdefault(modified_at, timezone.now()) queryset type(self).objects.filter(pkself.pk) if extra_filter: queryset queryset.filter(**extra_filter) updated queryset.update(**values) if updated ! 1 and extra_filter: current type(self).objects.filter(pkself.pk).values(status).first() logger.info( SafeUpdateGuardMiss: %s row %s extra_filter%s current_status%s loaded_status%s ..., ... ) if refresh: try: self.refresh_from_db() except type(self).DoesNotExist: pass return updated 1safe_update()用帶可選extra_filter的 UPDATE 實現(xiàn)帶守衛(wèi)的原子更新返回是否恰好更新 1 行。守衛(wèi)失敗時如extra_filter指定的狀態(tài)已被他人改變會記錄SafeUpdateGuardMiss日志便于排查競態(tài)。默認在更新后refresh_from_db()同步內(nèi)存態(tài)。update_and_requeue(**kwargs)safe_update的特化版本固定extra_filter{retry_at: self.retry_at}——即只有租約未變時才能更新再次確認所有權(quán)后才允許改寫狀態(tài)。runner 中大量使用它推進 Crawl/Snapshot 的調(diào)度狀態(tài)見 runner.py。save() 覆蓋runner 進程外的保存告警ModelWithQueue.save()覆寫了 Django 的save()實現(xiàn)了一個運行期安全網(wǎng)當非新增行在非 ORCHESTRATOR編排器進程中被save()時會通過棧幀回溯利用MODULE_PATH/PACKAGE_ROOT/REPO_ROOT定位調(diào)用者跳過本模塊與 site-packages 幀記錄警告日志指明調(diào)用方文件與行號。其意圖是隊列行的狀態(tài)變更應(yīng)當由 runner 統(tǒng)一管理直接在任意代碼里save()可能繞過租約守衛(wèi)造成狀態(tài)錯亂該告警幫助開發(fā)者盡早發(fā)現(xiàn)這類越權(quán)寫庫。子類可通過warn_on_save_outside_runner False關(guān)閉Binary 模型即如此因為它由BinaryService同步驅(qū)動安裝流程見 machine/models.py。輔助方法status_counts / extend_choices / StatusField / RetryAtFieldstatus_counts(querysetNone, statusesNone) - dict[str, int]類方法按狀態(tài)統(tǒng)計行數(shù)返回如{queued: 12, started: 3, ...}用于調(diào)度概覽與 admin 面板展示。extend_choices(base_choices)一個類裝飾器工廠把基礎(chǔ)枚舉與附加枚舉合并生成新的TextChoices供需要擴展狀態(tài)集的模型使用。StatusField(**kwargs)/RetryAtField(**kwargs)類方法字段工廠以默認字段定義為基礎(chǔ)合并覆蓋參數(shù)子類聲明字段時直接復(fù)用Crawl、Snapshot、Binary 三處均如此。子類化實踐Crawl、Snapshot 與 Binary從倉庫看ModelWithQueue被三個模型繼承展示了默認協(xié)議復(fù)用 按需定制的完整模式Crawl 與 Snapshot直接復(fù)用默認四態(tài)Crawl 與 Snapshot 都按默認協(xié)議定制# crawls/models.py class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): status ModelWithQueue.StatusField( choicesModelWithQueue.StatusChoices, defaultModelWithQueue.StatusChoices.QUEUED, ) retry_at ModelWithQueue.RetryAtField(defaulttimezone.now) StatusChoices ModelWithQueue.StatusChoices INITIAL_STATE StatusChoices.QUEUED ACTIVE_STATE StatusChoices.STARTED FINAL_STATES (StatusChoices.SEALED,) FINAL_OR_ACTIVE_STATES (*FINAL_STATES, ACTIVE_STATE)Snapshot 結(jié)構(gòu)相同core/models.py并額外定義RUNNABLE_STATES (QUEUED, STARTED)、OPEN_STATES等派生集合。兩者共享默認四態(tài)queued → started → sealed可被paused中斷。Binary完全重定義狀態(tài)機Binary 演示了最激進的定制——它只用兩態(tài)class StatusChoices(models.TextChoices): QUEUED queued, Queued INSTALLED installed, Installed status ModelWithQueue.StatusField(choicesStatusChoices.choices, defaultStatusChoices.QUEUED, max_length16) INITIAL_STATE StatusChoices.QUEUED ACTIVE_STATE StatusChoices.QUEUED # 活躍態(tài)即 QUEUED FINAL_STATES (StatusChoices.INSTALLED,) warn_on_save_outside_runner FalseBinary 的模塊 docstring 說明了設(shè)計意圖安裝流程是同步的queued → installed轉(zhuǎn)換失敗時保持queued并把retry_at順延以便稍后重試數(shù)據(jù)庫行是唯一的生命周期狀態(tài)worker 中斷后無需在內(nèi)存中二次協(xié)調(diào)狀態(tài)機。它還演示了update_and_requeue(retry_atNone, statusINSTALLED)的收尾用法machine/models.py。運行器中的真實調(diào)用鏈runnerservices/runner.py是把這套協(xié)議串起來的消費端其處理每個到期任務(wù)的標準流程是選行通過retry_at__ltenow類查詢等價于get_queue取出到期任務(wù)。認領(lǐng)調(diào)用crawl.claim_processing_lock(lock_secondslock_seconds)L1361、L1395或Snapshot.claim_for_worker(snapshot, lock_secondslock_seconds)L1448原子獲取租約失敗即返回False說明該行已被其他 worker 認領(lǐng)。執(zhí)行與續(xù)租處理期間用crawl.update_and_requeue(statusSTARTED, retry_atnow timedelta(secondsACTIVE_STATE_LEASE_SECONDS))續(xù)租L1352把租約從 60 秒起步按需延長。收尾終態(tài)任務(wù)如SEALED的 Crawl在claim_for_worker成功后執(zhí)行cleanup_runtime()最后update_and_requeue(retry_atNone)清空調(diào)度時間使其退出隊列L1405-L1411。值得注意的邊界處理當 Snapshot 的租約已過期但關(guān)聯(lián)的abx-dl鉤子進程仍在運行process_set.filter(statusrunning)時runner 不會啟動第二波處理而是update_and_requeue(retry_atnow lock_seconds)重新排隊L1471-L1479保留快照級的所有權(quán)邊界防止重復(fù)執(zhí)行。測試驗證與可觀測性倉庫測試對這套協(xié)議有直接覆蓋可作為理解佐證test_crawl_runner.py 斷言snapshot.claim_processing_lock(lock_seconds60) is True驗證單 worker 場景下的認領(lǐng)成功路徑同文件 L414 使用crawl.update_and_requeue(statusQUEUED, retry_attimezone.now())構(gòu)造調(diào)度場景驗證重排隊邏輯。此外認領(lǐng)失敗與守衛(wèi)未命中都會留下可觀測日志SafeUpdateGuardMisssafe_update 守衛(wèi)失敗以及save() outside runner process告警配合status_counts()的狀態(tài)分布統(tǒng)計可以在運維層面快速定位任務(wù)堆積在哪一態(tài)、是否被越權(quán)改寫。小結(jié)archivebox.workers.models用約 230 行代碼給出了一個干凈、可復(fù)用的數(shù)據(jù)庫級任務(wù)隊列協(xié)議以statusretry_at兩個索引字段為存儲以條件 UPDATE 影響行數(shù)判定實現(xiàn)原子認領(lǐng)與安全更新以pause/resume與RETRY_AT_MAX實現(xiàn)暫停語義以ACTIVE_STATE_LEASE_SECONDS維持活躍任務(wù)租約。Crawl、Snapshot、Binary 通過繼承與覆蓋各取所需runner 則作為唯一合法的狀態(tài)遷移入口消費隊列——這套mixin 管協(xié)議、模型管遷移、runner 管執(zhí)行的設(shè)計是 ArchiveBox 多進程調(diào)度一致性的基石。【免費下載鏈接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...項目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考