From cfaa67e9881364a1517beca8c81ce25da9d371da Mon Sep 17 00:00:00 2001 From: Shinya Okano Date: Tue, 13 Oct 2026 01:35:27 +0900 Subject: [PATCH] Add --max-tasks-jitter to spread out worker restarts Under --workers N --continuous --max-tasks K the workers start together and reach K at about the same time, so they restart together and the host briefly has no worker. Each worker now adds a random number from 0 to J, drawn once at start, to --max-tasks, as gunicorn's max_requests_jitter does. The draw is made in the worker, not the supervisor, so every worker and every restart gets its own, and it comes from SystemRandom so a project that seeds the random module does not make the draws alike. The limit drawn is printed at startup and carried as max_tasks on the Worker started record. The default of 0 leaves --max-tasks as it was, and the option is ignored without --max-tasks. Part of #64. --- CHANGELOG.md | 12 +++ README.md | 20 +++- .../locale/es/LC_MESSAGES/django.mo | Bin 10101 -> 10558 bytes .../locale/es/LC_MESSAGES/django.po | 49 +++++---- .../locale/ja/LC_MESSAGES/django.mo | Bin 10580 -> 11083 bytes .../locale/ja/LC_MESSAGES/django.po | 49 +++++---- .../locale/pt_BR/LC_MESSAGES/django.mo | Bin 10021 -> 10458 bytes .../locale/pt_BR/LC_MESSAGES/django.po | 42 +++++--- .../locale/zh_Hans/LC_MESSAGES/django.mo | Bin 9304 -> 9732 bytes .../locale/zh_Hans/LC_MESSAGES/django.po | 44 ++++---- .../management/commands/run_database_tasks.py | 30 +++++- tests/test_commands.py | 96 ++++++++++++++++++ tests/test_logging.py | 14 +++ 13 files changed, 282 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91ea5a6..b352337 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ ### Added +- **`--max-tasks-jitter J` on `run_database_tasks`.** Each worker adds a + random number from 0 to J, drawn once when it starts, to `--max-tasks`, + the way gunicorn's `max_requests_jitter` does, so the workers of + `--workers N --continuous --max-tasks K` no longer all reach K and restart + at the same moment. The draw is made by the worker, so every worker and + every restart gets its own, and from the operating system's random source, + so a project that seeds `random` does not make them alike; the limit drawn is printed at startup and + carried as a new `max_tasks` field on `Worker started`. Without + `--max-tasks` the option is ignored. The default is 0, which leaves + `--max-tasks` exactly as it was, and the only change to existing output is + the added log field, so existing projects need no changes. + ([#64](https://github.com/tokibito/django-database-task/issues/64)) - **A summary of the `--workers` supervisor's state in its log records.** When every worker is exiting with an error within ten seconds of starting at once, the supervisor logs one `Every worker is failing to start` diff --git a/README.md b/README.md index 4746b50..e0a38fe 100644 --- a/README.md +++ b/README.md @@ -256,6 +256,7 @@ python manage.py run_database_tasks [options] | `--continuous` | Keep polling even when no tasks | | `--interval` | Polling interval in seconds (default: 5) | | `--max-tasks` | Maximum number of tasks to process (0=unlimited) | +| `--max-tasks-jitter` | Add a random number from 0 to this to `--max-tasks`, drawn once per worker at start (default: 0). Ignored without `--max-tasks`. See [Running several workers](#running-several-workers) | | `--source` | Where to look for tasks: `auto` (default), `db`, `broker` or `both`. See [Task sources](#task-sources) | | `--wait-time` | Seconds to wait for a broker message before looking again (default: 20) | | `--max-messages` | Maximum number of broker messages to receive at a time (default: 1) | @@ -766,7 +767,24 @@ Shutdown complete: every worker exited. time. - **`--max-tasks`** goes to each worker. With `--continuous` that is gunicorn's `max_requests`: a worker exits after that many tasks and is - replaced, which bounds a leak in task code. + replaced, which bounds a leak in task code. The N workers start together + and process at similar rates, so on their own they reach the limit at + about the same time and restart together, leaving the host without a + worker for that moment. Add `--max-tasks-jitter J`, gunicorn's + `max_requests_jitter`, to spread them out: each worker draws a number + from 0 to J when it starts, a restarted one included, and stops after + `--max-tasks` plus that many tasks. A jitter of a tenth of `--max-tasks` + or more is a reasonable start: + + ```bash + python manage.py run_database_tasks --continuous --workers 4 \ + --max-tasks 1000 --max-tasks-jitter 100 + ``` + + The limit a worker drew is printed at startup + (`Max tasks: 1037 (1000 + jitter 37 of up to 100)`) and carried as + `max_tasks` on its `Worker started` log record. A worker run by a process + manager rather than `--workers` takes the option the same way. - **Signals.** `SIGTERM` and `SIGINT` (Ctrl-Break too on Windows) are handled by the supervisor, which forwards them to every worker and waits for the workers to finish their running task, within `--shutdown-timeout` if one is diff --git a/django_database_task/locale/es/LC_MESSAGES/django.mo b/django_database_task/locale/es/LC_MESSAGES/django.mo index c19ee004266032c269248e6ff4b8ae1688d766b2..8b2f083081c5dd1faf39fa22506509c91647bb97 100644 GIT binary patch delta 2092 zcmY+^TWl3Y7{KulsI|3dX%SQkbQCYeUZGMHilrh-Pa+nH^56q5J=2~|&+gXUJ(QTp z(FfxtS`&kb#6$&s@ls6ai=ZiuBucy`;)A!OF)^BmK@$_T##H=&yK7S?nf_+x>|DON z^!J^U_mrpWE1nlpj=YjwTqt7ky$b%2`YS~yaTi{JOD+~!j7`{x8*n4ui|g#%*9Ce|e;dJ4~S=l_brQ}`xf8SG*^I&crN>+%F{!DmrV zY^&Ry=_}px&vM8`ytt zp&VuZ7*jsP8}U1w#u}c)D4xT+ao;MD?RXpy;|w)j!rV3KWr|QE8$;G0$M84&1TWwzI=`Y4 zo$ky7*QJ3yj(SALu@B29C|pEg5nt&MRO73-5_N(pT)coCK|TBPH>BUqWY(}MeI=`r zQyyo7O*f?^7g>v(T=M+1tP5ZmX(RWN>-7CsQRpLU(#tbHEuzBs$wazpeLkA1 z$=k?!IX9B^4ryXDlcYyFKix)UJ$Wm+i`+xjC)mpT!gEdju|RSwHClZIcWOhH3~~jM zt=@uq=6X@rk^f5^{an(^th=a3z)2-{iPt=r^eDT@I)C!tPq)62e6u!?+Q}8<9C-&> zZ@DJDoIH-?73lu}z16pn?;vwUk~^UHDM`Kgym3tE`1!yye#zTsLo*ms?l5s+;({BM zwb0t?IR{!}CmN1gOg?l*eG~Wt){GXc&sP+JaG0)9?1V9&L_yXPrA;=^C_|PnLz54T zAH>Ef7ENfKfkM*1=z1<~aWZV1=^OHckWQm+ys8kC;`s|OO?f-$l#1~t)3J7M;=TzpDv#CT{S#E+TA_yahAYB9xBNs&y1PyK^ zs7)dfNQ5C0go`pHBEl%7C>KIxRAv!s-{1THTZ9*mpZ~q@Kiqrnxeo1{-V+%2gk3XQ zJ&{BlTV!?w`@;A`%U)vU#T<;ndQ8QGSb%3S2k)UD-(wlZhMTow6YBXv)N`*e1Am~$ zEMN%{W-A%U!pm5Joj8ij#hRBo54K<&{WkRBMSP6Cco@OvI0ASn!q_s!FTTXA6!E} zI>xy#69?&+;7k0Cm-xOtk9B6)!t%1|wU4nY?|+9-PNS>|qvj_y|dc z4Xkjs>IrHAFOa$0TVzf)iF(g34Cq5)q$Q6DvLyRil*u^asy zk>{A9z(yKMr$VQ3nw$DephraPIr!OOI;A=K7PVm*Gv#aNW#Y(b!$i`xwBM}1%h zqga7=owMf|i#LD?gJLG zfwRhh$2A?7(9cC~utL;_8c-8Cip(#VK4n7ps;Pi^=Yr;Egh!0T1dDm=e|H3&{Um$A zBc|$B0hSPKZSWtHW?w{v5^5yKe1x_olTg;wG)bjaEu0|Jmah(tlNQW|7Nl({Ahei7 zzQ3SVGNS6#nJsmH-Gar0s$C~$JE5#;PiqM^ZA%@YH0!vku9WI3LQN+?O$C@k>>{*K nt035JD<^7*9Yh{cNc?NLT\n" "Language-Team: Spanish\n" @@ -135,31 +135,42 @@ msgstr "Número de tareas a procesar de una vez (predeterminado: 1000)" msgid "Show what would happen without changing anything" msgstr "Muestra lo que ocurriría sin cambiar nada" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "Ejecuta las tareas encoladas en la base de datos" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "Nombre de la cola a procesar (todas las colas si no se indica)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "Nombre del backend (predeterminado: default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "Modo continuo (sigue sondeando aunque no haya tareas)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "Intervalo de sondeo en segundos en modo continuo (predeterminado: 5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "Número máximo de tareas a procesar (0=ilimitado, predeterminado: 0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"Suma a --max-tasks un número aleatorio de 0 hasta este valor, sorteado una " +"vez cuando el worker se inicia, para que los workers iniciados a la vez no " +"alcancen el límite al mismo tiempo. Se ignora sin --max-tasks " +"(predeterminado: 0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -171,7 +182,7 @@ msgstr "" "datos, y 'auto' (predeterminado) significa 'both' cuando el backend tiene un " "broker del que recibir y 'db' en caso contrario" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -180,12 +191,12 @@ msgstr "" "Sustituye a --interval como espera inactiva al recibir de un broker " "(predeterminado: 20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "" "Número máximo de mensajes del broker a recibir de una vez (predeterminado: 1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -196,18 +207,18 @@ msgstr "" "se reemplaza; sin él, el comando termina cuando todos los workers lo hayan " "hecho (predeterminado: 1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " "database connection of its own (default: 1)" msgstr "" -"Ejecuta las tareas en esta cantidad de hilos dentro del worker. El código " -"de las tareas debe ser seguro para hilos. El worker sigue consultando una " -"vez por intervalo; cada hilo tiene su propia conexión a la base de datos " +"Ejecuta las tareas en esta cantidad de hilos dentro del worker. El código de " +"las tareas debe ser seguro para hilos. El worker sigue consultando una vez " +"por intervalo; cada hilo tiene su propia conexión a la base de datos " "(predeterminado: 1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -216,7 +227,7 @@ msgstr "" "SIGINT antes de forzar la salida (0=esperar indefinidamente, predeterminado: " "0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -224,7 +235,7 @@ msgstr "" "No instala manejadores de SIGTERM/SIGINT; el proceso termina de inmediato, " "incluso con una tarea en ejecución" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -233,7 +244,7 @@ msgstr "" "planificador de trabajos distinga una ejecución vacía de una real " "(0=terminar normalmente, predeterminado: 0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/locale/ja/LC_MESSAGES/django.mo b/django_database_task/locale/ja/LC_MESSAGES/django.mo index 532ff556aa722f44008c53d85441c3131814c059..b3fc2869e0a0bb9286afa745f73f7c0a73a1b6c2 100644 GIT binary patch delta 2099 zcmY+_eN2^A9Ki7(WJ+3snoufnOiFy=MQ&;-rDzu1^@m#NdRx13FJ8vw-gWO~&{kcZ zn=S3i#cPu?_BU1%_j7w`bS zjQ%+yTd@zT@OQLu6|*)tfI86;tmOG3cZqBOAO3?;)Q*w4BKi0!GPYbmA71qQ7j?qE zTz3IAs1tYr@56Im{Z(8<`8ThelgI8-UW}LZ`B%uaV?WC>xQ^|pz}?8M%YJ+myHHQ% zdt^~^8BgFfT#Rol5ShYWWYgprvtyC+A?gAKk+sR!$lT>SOzA*BlF?APc{;o z9&&LmvSv9!=k?RP-Z9o8WkfYGPv8F>GPQ&zy?oQtZ4_1zI%DpZ^ZJ0ZvaPt5SWmFq zPGYy6^qA*OwveeLHWPXbn+TnJDbIJ_03T|~CUl!u6HgL1lOeN|&<~P+3%G~E2033c za1vMJ|D<33cqmX8cKSyeni{A}I>VNk+C9-&oK7tb$-;1~IXQg-rX(1u4>U)T8%)KDvlTN3 za#Pc$XKbsV$=gY`FPJ~cKY2+`v{wT*T2@8i7&p)bo7k(rgh9UCUg9RZPDwU%y5rwbq38R#rOv8_(q>T|~JbEkFds@z}qIxP>{ d*5EIDPfnaZYFj;ay1To$Ae+7Td132<{{RR(cRT<9 delta 1611 zcmZY9OGs2v7{KvQv&Kg`rrG$YX6#{3X+EaXa;&Ve90*FCltPH0gi;VG(nWYta8YIj z8MzW#CS`$b3KvDS$q9V>DQ@5S+tQnXCOgT=TU!|26&EX9jhjKf%nA8{|HxI~)qDC+M=QGXZ3jrapo zMJ&mT7gI2?7?wlE)q#_e%OF3skfsKFXK}j!jssXC~^SDFo^CXkuE%pJ{(7# z=q#3C4y)PC_!8k_frjg-KkQo}vKn6@eaSQi@Qcxx>`XX@stAhY<5Pms-! z$7#+{Jx5)@1TuG-MCK%)QTv%gO9zT$|Ft0r=~FyLKXM9Ej@nTO*(5oFI#8o&Z$j;- z&9rxz_HJVzhGIkpkTGQ{U4%dRmd^g`ie2pY2`s}ZylH%gHPpRiP#d;k27W?K)?c_E zGilVLXf(FsIQ4Fv#F|VemkKzK_0)?{pRdbe|2K2dM8hmz!%pmEUn97Hl~~JY{n(HC zynn68F03*3;&JLPF~CZ_BtnRKo!2>{0o0v)gw^=Wm}vQ&71bGA@H`LnpbjvPIT+v{ zn<2-LzNJNt5$VMe9Aa4$v`=kto*(8N+ev)_$un_nViD*;J-TzKxno`6;wu+7umHPq zMOx|b0qU80^PIEFMO{G<^@m~O8FRl2wd32U&)qTY!>C*R!rXsr?oT^)OJ>XibLK{p z-{ zsAxu$xb{0+VeabDloR^4?5j(G9o}|0{$x{T|2@#x@jJRF*-6K&?d)i;O4UH!N diff --git a/django_database_task/locale/ja/LC_MESSAGES/django.po b/django_database_task/locale/ja/LC_MESSAGES/django.po index 007e2a2..bcb0c63 100644 --- a/django_database_task/locale/ja/LC_MESSAGES/django.po +++ b/django_database_task/locale/ja/LC_MESSAGES/django.po @@ -7,7 +7,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2025-12-27 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Japanese\n" @@ -129,31 +129,41 @@ msgstr "一度に処理するタスク数 (デフォルト: 1000)" msgid "Show what would happen without changing anything" msgstr "何も変更せずに、何が行われるかを表示" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "データベースのキューに登録されたタスクを実行" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "処理するキュー名 (指定しない場合はすべてのキュー)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "バックエンド名 (デフォルト: default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "継続モード (タスクがなくてもポーリングを続ける)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "継続モードのポーリング間隔（秒）(デフォルト: 5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "処理する最大タスク数 (0=無制限、デフォルト: 0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"--max-tasks に 0 からこの値までの乱数を加える。乱数はワーカーの起動時に 1 回" +"だけ引くので、同時に起動したワーカーが一斉に上限に達することがなくなる。--" +"max-tasks がないときは無視される (デフォルト: 0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -165,7 +175,7 @@ msgstr "" "バックする。'auto' (デフォルト) は、受信できるブローカーがバックエンドにあれ" "ば 'both'、なければ 'db' になる" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -173,11 +183,11 @@ msgstr "" "再確認するまでブローカーのメッセージを待つ秒数。ブローカーから受信する場合、" "アイドル時の待機として --interval の代わりに使われる (デフォルト: 20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "一度に受信するブローカーメッセージの最大数 (デフォルト: 1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -187,18 +197,17 @@ msgstr "" "このコマンドのコピー。--continuous 指定時は終了したワーカーを再起動し、未指定" "時は全ワーカーが終了するとコマンドも終了する (デフォルト: 1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " "database connection of its own (default: 1)" msgstr "" -"ワーカー内でこの数のスレッドでタスクを実行する。タスクのコードはスレッドセーフである必要がある。ポーリングは引き続き " -"interval ごとに 1 " -"回で、スレッドごとにデータベース接続を 1 本持つ " -"(デフォルト: 1)" +"ワーカー内でこの数のスレッドでタスクを実行する。タスクのコードはスレッドセー" +"フである必要がある。ポーリングは引き続き interval ごとに 1 回で、スレッドごと" +"にデータベース接続を 1 本持つ (デフォルト: 1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -206,7 +215,7 @@ msgstr "" "SIGTERM/SIGINTを受信してから強制終了するまで、実行中のタスクを待つ最大秒数 " "(0=無期限に待つ、デフォルト: 0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -214,7 +223,7 @@ msgstr "" "SIGTERM/SIGINTハンドラーを設定しない。タスクの実行中でもプロセスは即座に終了" "する" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -222,7 +231,7 @@ msgstr "" "タスクを1件も処理しなかった場合にこの終了コードで終了する。ジョブスケジュー" "ラーが空振りの実行を区別できるようにする (0=通常どおり終了、デフォルト: 0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo b/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo index 275e84e367352efaabc34d9aa5bf5111315b1067..2628a31421e9e7b39cc229f5f91ebbe68c48e829 100644 GIT binary patch delta 2051 zcmY+_eP~r>9Ki9PwsdBtGiT+pJ-M{4y|$}c&CF@d=7MxYN)*)7&e40fd(N5Xoa+@K zF8xttDS{ZiSx_rNMU2*8Ym_yjHxx3&SRh2?A40E$gwV_Pcka2c9(eBOInO=MbAHe7 zdCtA>JUBgAI=`suMWGB(Z=&XBi?}#h#Rp}}9FdQ35UZxt!Yw=@>f@fR+}`SV2fp}}YHI1-a&t||}g!Wyo(qQNYl z#J%`94$K$17mwoI_zOOb%|vbRIBKA$v6JVETqDv08T=c4)PebHMV8l+ zU(|pz3(5=VM-AXnycs`C`=@a!*FUG%)eFfk*L8SCpZ^_==kXZJGT22rI&d43b$Jfg z;sMlMIfX1r&fpt(9_#R_>qP#+SCFL1%S6W_F03HiT~#=D#> zvLm11a{U7P@K5ZX#Phb%*p+TQXI81T?b#FgFo#z7Hg7e7#>)4EM|!t1`%L*QIE9+QpGq{=(fA{^lW;ZSkdGC32#0Y3n{WluYQUTMU``2913!qr;d^)) z-)!I<;9=6D^K|i{;~zuaqGz!eOZ(G~)5tlLvv>?IpswuTNCWWSyUUYr`G8EUrl2ZRg1oMGs_iRXxezg zlGSNoT6E?5BeR;S`>uyi3rS1LJ*gEo(CDQ0QoE?S<(j2do}Z*8gIZ=&b+s+j`>B;; zXmA9QAEdRYdwx4rPkW^-uQX7%M6)xO%8m%fAxS$VMYmFat+T8o4_A(O)=~G+(jb@M;IE#HjKF?3euja+~qpVvcu)h z&^UqdgV@-7-h{3l$|mFUIWI?FoJ`whHtq0(kU_;7l^0EMrELroeKhCOkI`c#>=5sqruu7{_H{TpTfO6uV}Z%?F`I8^*3Bt^(8S za`%{>1x<*lEW+gc+)&Q8m{A*AzN{=aO263Z1d$mo_%ZlXV>~g=`NP&@f@EE;v4Qd2 g$Rne!HyWht(Ad(aQ3YnwJtH3l2#TSP(%8Q4s1waZvD~i}X}f zluw0}5rPmH3X$|uAr(SlnH5A7V$lUkG5ddKCVFtg@;9@c*|YarOGkF!*co^e7ka@M zwL}_mFvRQt-VWslBQM-634IuewU~~3un?Qjhj%a+-(nfYc+47c7wY|f)O*9Y0)Jqf zS-?`}nZ+}agXgdU+i?ili}~j}FV#nFhkC9bm*7|m`CrM!G!vgOlY*Va81lLuuVOiVK&3n@ z&1@Au#6q0F&6;PmI^2(0oU{kaxqm@Jus2IicQTiUW!#s#o(gcG9S>k8j^Pmu^O|Mj zG1QK)AiuH!)XqooD^B4gjtD$#!bk@jaMkQ7j)y(UN|07_whC7uY zl+eyL5G4eq4eng0EF$I-8cL9tpjCn;Qdu-qXJtUcLy&1(uN#IkLDL6Qz8cj#`rnbt z_h*f*%qYb=b*cN?wOCB(|3eP3kx({ji5fzKKayb6s!U2Xy%)@2C2E8?*cQ}Z%{D>@ mwF*K_Q%+P9n}`BJo&9g*bD=tGR1xZx-e}K*=-#fRg4n-zgO6wc diff --git a/django_database_task/locale/pt_BR/LC_MESSAGES/django.po b/django_database_task/locale/pt_BR/LC_MESSAGES/django.po index 8ca5a27..4b42e28 100644 --- a/django_database_task/locale/pt_BR/LC_MESSAGES/django.po +++ b/django_database_task/locale/pt_BR/LC_MESSAGES/django.po @@ -6,7 +6,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2026-09-30 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Brazilian Portuguese\n" @@ -132,31 +132,41 @@ msgstr "Número de tarefas a processar de uma vez (padrão: 1000)" msgid "Show what would happen without changing anything" msgstr "Mostra o que aconteceria sem alterar nada" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "Executa as tarefas enfileiradas no banco de dados" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "Nome da fila a processar (todas as filas se não for informado)" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "Nome do backend (padrão: default)" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "Modo contínuo (continua consultando mesmo sem tarefas)" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "Intervalo de consulta em segundos no modo contínuo (padrão: 5)" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "Número máximo de tarefas a processar (0=ilimitado, padrão: 0)" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"Soma a --max-tasks um número aleatório de 0 até este valor, sorteado uma vez " +"quando o worker inicia, para que workers iniciados juntos não atinjam o " +"limite ao mesmo tempo. Ignorado sem --max-tasks (padrão: 0)" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -168,7 +178,7 @@ msgstr "" "e 'auto' (padrão) significa 'both' quando o backend tem um broker do qual " "receber e 'db' caso contrário" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -176,11 +186,11 @@ msgstr "" "Segundos de espera por uma mensagem do broker antes de verificar novamente. " "Substitui --interval como espera ociosa ao receber de um broker (padrão: 20)" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "Número máximo de mensagens do broker a receber de uma vez (padrão: 1)" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -191,7 +201,7 @@ msgstr "" "que sai é substituído; sem ele, o comando termina quando todos os workers " "terminarem (padrão: 1)" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " @@ -202,7 +212,7 @@ msgstr "" "por intervalo; cada thread tem a sua própria conexão com o banco de dados " "(padrão: 1)" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -210,7 +220,7 @@ msgstr "" "Tempo máximo, em segundos, de espera pela tarefa em execução após receber " "SIGTERM/SIGINT antes de forçar a saída (0=esperar indefinidamente, padrão: 0)" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" @@ -218,7 +228,7 @@ msgstr "" "Não instala manipuladores de SIGTERM/SIGINT; o processo é encerrado " "imediatamente, mesmo com uma tarefa em execução" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -227,7 +237,7 @@ msgstr "" "agendador de jobs distinga uma execução ociosa de uma real (0=sair " "normalmente, padrão: 0)" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo index 1920335620dbf67f9475b39947231f26210e0ffb..034f0c42f6813bbac59e80e715082c35e6dfeabf 100644 GIT binary patch delta 2041 zcmY+^eN2^A9Ki8kNk;jGFIa)-V_IQ&fgm;CF~g8tvsc&HY8UUt%edUT-Fu1rBQG^u zyg9IqM7VR+bVCYc(Twz7@SmE^><|4TrkksI?*(eb+FI+MvhVMCj)iCY@j2&kp7WfS z-*d++4)2NHSd`HvluBY5;lEqNi}4KpQ1;Fi8O0ay0nEBrWG?383S5U9@D+Rlk6;mw z;dY$HIxNa`p6^9{ZWvdKMCB@#MRfdyOK{G8BJF75dpLlsNlNc`9$b$LX}^jV`tSm_ z;2T&uM`S1V;}-l4-@yV_ZQ%gwLchln-Y+swWD}I)-xxrB&_7>fDV|2=mJ3*lmyEYi z7hJl)*+2#A0;+I1o;Us1Fo*W9rk$C^>C(={EBgL%Dy{e_+p=&y$5D*?kW-gt+=?Bj zw{jWTlw83Rcms3s?ZqPh;Bn;8q?^^TN%;hI180!E$vI^0auK6C(GOJgMK@8~KcgP; zpT^8=5pG9vP$yc2I&mTDi#D77a@2YDn*M#J-);2cOLvJh;&+U@w3PGLr0V96PB?@Q zV*)qgkEkb+^Pq?epF_QxgZMh0MK&+l4~gVqK9W?j3$?$_cnA;CK8kwMS=@)_RBsOF zZ&5ixhaTY=K8DxvYn(-=55L7SELq0&U>FZ#Kk9Rk)! zTc{h(jjnKZns2N|PDf(K&+%p27f~0MXF2~bN7g7+s5^ZVN3jK`uwo^3Ea5nHfpOI2 z9Y?*QxvNCBVRSJS?RXJ&MX%u?)}SsVoAdk$3sA?+d&C*H#JCbkY$-HuGyTt_j;}K9 zYV-df>Vl(4?nLDSGa%-4$Z0cR*z|vcI?+|rzGj?2eeoYiPRV9M%aaavB5Rc!vQJ9u zx}kickk~;i)bC$2dpDs)lWk_1MPm)2D<*gB&j;+R--+vp4TL707S>`HJ=&Q?bD@MN zBQ$@u6Z#3dc)#$u+IItXR_boo63=RfDHd|2_7BolB%kDQLX+xt;cEXYd>HoW@at>m z4Yz6kSlu3=pYaw#-*5l=@f&3SEp9u2|Hb053?NbGBRX&f>t0HvD|*Y74o|G`|Ria zwe_|1MeJz{t=$I#!4MBM)<*Jt!G_4p1z7nWZ;iXbAK7FTuNf|$omdc!_1!TbIntVH z{U|xolpN`2BdO-0^uWnfQ#|=q+tjfyCVCD}^>n0T=h9t6)A9CHtS|NDsbnHHIo3PT z*Oux$lWOftA3K|FO{6+LWRmn}A2TeO7)y1w)9(ymlK-FD)7O?9Ig*U`I3p%UM|mZS zm^$8hv+3P*tUulUermAijy38`V~L5bQ`7P0sq@{FgD0osJ;l@Umf>Hr>*xOqZVp#1 delta 1611 zcmZY9TS$~a6u|MLZEk6~UP|-6b~DRdZ)LTbnpQ3d1xou8l^~@Eg%6d1v?L`YNe~4Y zVbDWRlm>#3SP2q=m4x*WbRi!K1WCI*6ruG$-%JnHz}er-e3zN;%*>vyc~u>H7w5Vp zlmL-RG=z)P<82o|l)MO$M9jxU7{CnNgT;6Z^RX9w_#VqJ#x2r}yHTGXLVfNfuEcK` zClZpR1tLr6$iZ`1foE_GnTymdG!F(bmiBS<;#nNVTeu&aBSk9l36`KIO5_yQpcltb zCpwJ}KQWnh zw8z|+jYG6ca2$VNC*#Z07&F5lpO;6w70II9#8SMAx*DI5$;(&l!C#nyJvR|5H8PDd>H-i$@4890pV@fzxqJi~n$$SnrP`yX1z@g)5puoD}WnY}cPUfN$!GZ&vk{e3v*t$4_`Z=g;nbcaeWl|j^jnplqsy@2{) z{R(qqqxC4Vai!gQ!S-KAy}!@42ki9`)Cs>p-7_EU^%>I+$*jG>%|hyol2KojW!t${ zKkC5M7=~p8NptQ6X9KU+%0IGN!O4Jf=aP~ny^__%%s8kTEh;m{(v56=o cbf^9)1yuCULrWF0hUoXW?|b^Y6Pu%d12#>KH2?qr diff --git a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po index 0035b39..379333c 100644 --- a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po +++ b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po @@ -6,7 +6,7 @@ msgid "" msgstr "" "Project-Id-Version: django-database-task 0.1.0\n" "Report-Msgid-Bugs-To: \n" -"POT-Creation-Date: 2026-10-02 05:07+0000\n" +"POT-Creation-Date: 2026-10-12 16:27+0000\n" "PO-Revision-Date: 2026-09-30 00:00+0000\n" "Last-Translator: Shinya Okano \n" "Language-Team: Simplified Chinese\n" @@ -123,31 +123,40 @@ msgstr "每次处理的任务数（默认：1000）" msgid "Show what would happen without changing anything" msgstr "显示将会执行的操作，但不做任何更改" -#: management/commands/run_database_tasks.py:63 +#: management/commands/run_database_tasks.py:70 msgid "Execute tasks queued in the database" msgstr "执行在数据库中排队的任务" -#: management/commands/run_database_tasks.py:70 +#: management/commands/run_database_tasks.py:77 msgid "Queue name to process (all queues if not specified)" msgstr "要处理的队列名称（未指定时为所有队列）" -#: management/commands/run_database_tasks.py:76 +#: management/commands/run_database_tasks.py:83 msgid "Backend name (default: default)" msgstr "后端名称（默认：default）" -#: management/commands/run_database_tasks.py:81 +#: management/commands/run_database_tasks.py:88 msgid "Continuous mode (keep polling even when no tasks)" msgstr "持续模式（没有任务时也继续轮询）" -#: management/commands/run_database_tasks.py:87 +#: management/commands/run_database_tasks.py:94 msgid "Polling interval in seconds for continuous mode (default: 5)" msgstr "持续模式的轮询间隔秒数（默认：5）" -#: management/commands/run_database_tasks.py:93 +#: management/commands/run_database_tasks.py:100 msgid "Maximum number of tasks to process (0=unlimited, default: 0)" msgstr "要处理的最大任务数（0=不限，默认：0）" -#: management/commands/run_database_tasks.py:100 +#: management/commands/run_database_tasks.py:108 +msgid "" +"Add a random number from 0 to this to --max-tasks, drawn once when the " +"worker starts, so workers started together do not all reach the limit at " +"once. Ignored without --max-tasks (default: 0)" +msgstr "" +"在 --max-tasks 上加一个 0 到此值之间的随机数，在工作进程启动时抽取一次，使同" +"时启动的工作进程不会同时达到上限。未指定 --max-tasks 时忽略（默认：0）" + +#: management/commands/run_database_tasks.py:119 msgid "" "Where to look for tasks: 'db' polls the database, 'broker' receives from the " "backend's broker, 'both' does the broker first and falls back to the " @@ -158,7 +167,7 @@ msgstr "" "消息代理再回退到数据库，'auto'（默认）在后端有可接收的消息代理时为 'both'，否" "则为 'db'" -#: management/commands/run_database_tasks.py:112 +#: management/commands/run_database_tasks.py:131 msgid "" "Seconds to wait for a broker message before looking again. Replaces --" "interval as the idle wait when receiving from a broker (default: 20)" @@ -166,11 +175,11 @@ msgstr "" "再次查看前等待消息代理消息的秒数。从消息代理接收时，取代 --interval 作为空闲" "时的等待（默认：20）" -#: management/commands/run_database_tasks.py:122 +#: management/commands/run_database_tasks.py:141 msgid "Maximum number of broker messages to receive at a time (default: 1)" msgstr "每次接收的消息代理消息的最大数量（默认：1）" -#: management/commands/run_database_tasks.py:131 +#: management/commands/run_database_tasks.py:150 msgid "" "Run this many worker processes from one command, each a copy of this command " "without --workers. With --continuous a worker that exits is replaced; " @@ -180,15 +189,16 @@ msgstr "" "continuous 时，退出的工作进程会被替换；未指定时，所有工作进程退出后命令即退出" "（默认：1）" -#: management/commands/run_database_tasks.py:143 +#: management/commands/run_database_tasks.py:162 msgid "" "Run tasks in this many threads inside the worker. The task code must be " "thread-safe. The worker still polls once per interval; each thread has a " "database connection of its own (default: 1)" msgstr "" -"在工作进程内用这么多个线程运行任务。任务代码必须是线程安全的。工作进程仍每个间隔轮询一次；每个线程有自己的数据库连接（默认：1）" +"在工作进程内用这么多个线程运行任务。任务代码必须是线程安全的。工作进程仍每个" +"间隔轮询一次；每个线程有自己的数据库连接（默认：1）" -#: management/commands/run_database_tasks.py:154 +#: management/commands/run_database_tasks.py:173 msgid "" "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT " "before forcing exit (0=wait indefinitely, default: 0)" @@ -196,13 +206,13 @@ msgstr "" "收到 SIGTERM/SIGINT 后、强制退出前等待正在运行的任务的最长秒数（0=无限期等" "待，默认：0）" -#: management/commands/run_database_tasks.py:163 +#: management/commands/run_database_tasks.py:182 msgid "" "Do not install SIGTERM/SIGINT handlers; the process is terminated " "immediately, even while a task is running" msgstr "不安装 SIGTERM/SIGINT 处理程序；即使有任务正在运行，进程也会立即终止" -#: management/commands/run_database_tasks.py:173 +#: management/commands/run_database_tasks.py:192 msgid "" "Exit with this code when no task was processed, so a job scheduler can tell " "an idle run from a real one (0=exit normally, default: 0)" @@ -210,7 +220,7 @@ msgstr "" "未处理任何任务时以此代码退出，以便作业调度器区分空转的运行与实际的运行（0=正" "常退出，默认：0）" -#: management/commands/run_database_tasks.py:184 +#: management/commands/run_database_tasks.py:203 msgid "" "Exit with this code when at least one task failed or could not be run. Takes " "precedence over --empty-exit-code (0=exit normally, default: 0)" diff --git a/django_database_task/management/commands/run_database_tasks.py b/django_database_task/management/commands/run_database_tasks.py index 88484a0..b45bcdb 100644 --- a/django_database_task/management/commands/run_database_tasks.py +++ b/django_database_task/management/commands/run_database_tasks.py @@ -1,6 +1,7 @@ import argparse import functools import logging +import random import socket import sys import threading @@ -98,6 +99,18 @@ def add_arguments(self, parser): default=0, help=_("Maximum number of tasks to process (0=unlimited, default: 0)"), ) + parser.add_argument( + "--max-tasks-jitter", + type=int, + default=0, + metavar="J", + help=_( + "Add a random number from 0 to this to --max-tasks, drawn once " + "when the worker starts, so workers started together do not " + "all reach the limit at once. Ignored without --max-tasks " + "(default: 0)" + ), + ) parser.add_argument( "--source", choices=SOURCES, @@ -214,6 +227,9 @@ def handle(self, *args, **options): threads = options["threads"] if threads < 1: raise CommandError("--threads must be at least 1") + max_tasks_jitter = options["max_tasks_jitter"] + if max_tasks_jitter < 0: + raise CommandError("--max-tasks-jitter must not be negative") backend = task_backends[backend_name] source = self._resolve_source(backend, options["source"]) @@ -226,6 +242,12 @@ def handle(self, *args, **options): worker_id = f"{socket.gethostname()}-{uuid.uuid4().hex[:8]}" worker_index = self._read_worker_index() + # Drawn here, in the worker, rather than by the --workers supervisor, + # so that every worker and every restart gets a value of its own; and + # from the OS rather than the random module's shared generator, which + # a project that seeds it at import would make draw alike everywhere. + jitter = random.SystemRandom().randint(0, max_tasks_jitter) if max_tasks else 0 + max_tasks += jitter if verbosity >= 1: self.stdout.write(f"Worker ID: {worker_id}") if worker_index is not None: @@ -243,7 +265,12 @@ def handle(self, *args, **options): f"Broker: {type(backend.broker).__name__} " f"(wait={wait_time}s, max_messages={max_messages})" ) - if max_tasks: + if max_tasks and max_tasks_jitter: + self.stdout.write( + f"Max tasks: {max_tasks} ({options['max_tasks']} + jitter " + f"{jitter} of up to {max_tasks_jitter})" + ) + elif max_tasks: self.stdout.write(f"Max tasks: {max_tasks}") if graceful: timeout_label = ( @@ -275,6 +302,7 @@ def handle(self, *args, **options): "queue_name": queue_name, "continuous": continuous, "threads": threads, + "max_tasks": max_tasks, **worker_log_fields(), }, ) diff --git a/tests/test_commands.py b/tests/test_commands.py index bfc4510..e36a606 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -1,6 +1,8 @@ """Tests for management commands.""" import os +import random +import re import signal import sqlite3 import subprocess @@ -156,6 +158,94 @@ def test_run_database_tasks_no_tasks(self): assert "No more tasks to process" in out.getvalue() +@pytest.mark.django_db +class TestRunDatabaseTasksMaxTasksJitter: + """--max-tasks-jitter: a random addition to --max-tasks, drawn per worker.""" + + def run(self, drawn=0, **options): + out = StringIO() + with patch("random.SystemRandom.randint", return_value=drawn) as randint: + call_command("run_database_tasks", stdout=out, **options) + return out.getvalue(), randint + + def test_the_drawn_value_is_added_to_max_tasks(self): + for i in range(5): + simple_task.enqueue(i, i) + + output, randint = self.run(drawn=2, max_tasks=1, max_tasks_jitter=3) + + randint.assert_called_once_with(0, 3) + assert "Max tasks: 3 (1 + jitter 2 of up to 3)" in output + assert "Reached max tasks limit: 3" in output + assert ( + DatabaseTask.objects.filter(status=TaskResultStatus.SUCCESSFUL).count() == 3 + ) + assert DatabaseTask.objects.filter(status=TaskResultStatus.READY).count() == 2 + + def test_a_draw_of_zero_leaves_max_tasks_as_it_is(self): + for i in range(3): + simple_task.enqueue(i, i) + + output, _ = self.run(drawn=0, max_tasks=2, max_tasks_jitter=3) + + assert "Max tasks: 2 (2 + jitter 0 of up to 3)" in output + assert ( + DatabaseTask.objects.filter(status=TaskResultStatus.SUCCESSFUL).count() == 2 + ) + + def test_the_default_is_no_jitter(self): + """Without the option --max-tasks is what it has always been.""" + output, randint = self.run(max_tasks=2) + + randint.assert_called_once_with(0, 0) + assert "Max tasks: 2\n" in output + assert "jitter" not in output + + def test_jitter_without_max_tasks_is_ignored(self): + """Unlimited stays unlimited, as with gunicorn's max_requests_jitter.""" + for i in range(3): + simple_task.enqueue(i, i) + + output, randint = self.run(drawn=1, max_tasks_jitter=5) + + randint.assert_not_called() + assert "Max tasks" not in output + assert ( + DatabaseTask.objects.filter(status=TaskResultStatus.SUCCESSFUL).count() == 3 + ) + + def test_a_seeded_random_module_does_not_make_the_draws_alike(self): + """A project that seeds random at import must not undo the jitter.""" + drawn = [] + state = random.getstate() + try: + for _ in range(2): + random.seed(0) + out = StringIO() + call_command( + "run_database_tasks", + max_tasks=1, + max_tasks_jitter=10**9, + stdout=out, + ) + drawn.append(re.search(r"Max tasks: (\d+)", out.getvalue())[1]) + finally: + random.setstate(state) + + assert drawn[0] != drawn[1] + + def test_jitter_must_not_be_negative(self): + with pytest.raises( + CommandError, match="--max-tasks-jitter must not be negative" + ): + call_command( + "run_database_tasks", + max_tasks=1, + max_tasks_jitter=-1, + stdout=StringIO(), + ) + + @posix_signals @pytest.mark.django_db class TestRunDatabaseTasksGracefulShutdown: @@ -1375,6 +1465,12 @@ def test_the_workers_exit_codes_become_the_commands(self): self.run_workers(code=1, workers=2, failed_exit_code=1) assert exc_info.value.code == 1 + def test_the_jitter_is_drawn_by_each_worker_not_the_supervisor(self): + with patch("random.SystemRandom.randint") as randint: + self.run_workers(workers=2, max_tasks=10, max_tasks_jitter=5) + + randint.assert_not_called() + def test_a_normal_exit_of_every_worker_exits_normally(self): self.run_workers(code=0, workers=2, empty_exit_code=4, failed_exit_code=1) diff --git a/tests/test_logging.py b/tests/test_logging.py index 199d313..c8d6e3c 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -275,6 +275,20 @@ def test_start_is_logged_with_fields(self, task_logs): assert record.source == "db" assert record.queue_name == "emails" assert record.continuous is False + assert record.max_tasks == 0 + + def test_start_carries_the_max_tasks_with_the_jitter_drawn(self, task_logs): + """The drawn jitter is otherwise invisible to a fleet's logs.""" + with patch("random.SystemRandom.randint", return_value=7): + call_command( + "run_database_tasks", + max_tasks=100, + max_tasks_jitter=10, + stdout=StringIO(), + ) + + (record,) = records_matching(task_logs, "Worker started") + assert record.max_tasks == 107 def test_a_failed_receive_is_logged_with_the_worker_fields(self, task_logs): class BrokenBroker(FakePullBroker):