amadan
Mail queue: nothing on the request path, a worker that survives, a close that ends
Astra on #20, round 2. P1: Send logged "queue full" on the request path, so a stalled log sink held the request, and saturation was something to time. Send now does no I/O at all: a drop is counted in an atomic and the worker logs the count. Every address's link is queued, permitted or not, so the fill level does not depend on admission, and a full queue answers exactly like an empty one. P2: a panic in the mailer killed the process. The worker recovers per message, logs, and keeps draining. P2: close waited 30 s and returned with the worker still blocked. Sends now run under the queue's context; past the drain deadline close cancels the send in flight, abandons what is left with a log line, and the worker exits. Tests: TestMailQueueSendNeverBlocksWhenFull (stalled log sink, twice the depth), TestMailQueueSurvivesAPanic, TestMailQueueCloseIsBoundedAndStops TheWorker (goroutine count back to baseline), TestAppCloseStopsTheMail Worker, TestMailQueueQueuesEveryAddress, and over HTTP TestAFullMailQueueAnswersLikeAnEmptyOne. Each of the first four fails with its defect put back.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+333
−44
internal/slides/app.go+1 −1internal/slides/mailqueue.go+110 −43internal/slides/mailqueue_test.go+180 −0internal/slidestest/signin_timing_test.go+42 −0
CI passed — run details
CI log
=== gate ===
=== rust: fmt
=== rust: clippy
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.17s
=== wasm: the artifact the host embeds
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.04s
Finished `release` profile [optimized] target(s) in 0.11s
core/target/wasm32-unknown-unknown/release/slides_core.wasm (1228229 bytes)
=== the embedded wasm must match a fresh build
=== rust: test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.04s
Running unittests src/lib.rs (core/target/debug/deps/slides_core-929bfceb2edd4a94)
running 30 tests
test check::tests::bare_text_at_fragment_level_is_reported ... ok
test check::tests::an_unknown_formatting_attribute_is_reported ... ok
test check::tests::an_element_inside_a_paragraph_is_reported ... ok
test layout::tests::aside_region_rect ... ok
test check::tests::the_empty_document_passes ... ok
test layout::tests::default_region_by_kind ... ok
test check::tests::what_the_ops_write_passes ... ok
test layout::tests::body_region_rect_narrows_for_an_aside ... ok
test layout::tests::heading_region_rect ... ok
test markdown::tests::inline_marks ... ok
test schema::tests::an_empty_map_reads_as_a_set_mark ... ok
test schema::tests::an_href_map_without_a_string_url_reads_as_no_link ... ok
test layout::tests::media_region_rect ... ok
test schema::tests::href_is_written_as_a_one_key_map_and_reads_back ... ok
test layout::tests::to_canvas_narrows_body_when_an_aside_is_present ... ok
test layout::tests::to_flow_classifies_by_position_and_kind ... ok
test layout::tests::to_flow_orders_y_then_x ... ok
test schema::tests::href_written_as_a_plain_string_still_reads_back ... ok
test check::tests::a_non_paragraph_child_is_reported ... ok
test markdown::tests::empty_input_is_no_paragraphs_and_lines_keep_blanks ... ok
test markdown::tests::marks_nest_and_links_carry_marks ... ok
test layout::tests::to_flow_is_a_no_op_when_already_in_order ... ok
test markdown::tests::paragraphs_split_on_blank_lines_and_list_markers ... ok
test check::tests::y_prosemirror_shapes_pass ... ok
test layout::tests::to_flow_keeps_a_concurrent_text_edit ... ok
test layout::tests::to_canvas_splits_a_shared_region_equally_and_sets_z_by_index ... ok
test view::tests::malformed_fonts_are_null ... ok
test view::tests::assets_are_typed_or_skipped ... ok
test view::tests::malformed_box_is_null_and_phone_keeps_only_typed_fields ... ok
test view::tests::malformed_background_is_the_default ... ok
test result: ok. 30 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.01s
Running tests/no_host_imports.rs (core/target/debug/deps/no_host_imports-184c0d8bf00e14f9)
running 1 test
test no_host_imports ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
Running tests/ops.rs (core/target/debug/deps/ops-256addce27f366d1)
running 12 tests
test add_block_with_markdown_produces_runs ... ok
test add_slide_then_view_lists_it ... ok
test columns_children_must_name_blocks_on_the_slide ... ok
test columns_children_have_one_parent_and_no_cycles ... ok
test ops_are_atomic ... ok
test layout_switch_assigns_boxes ... ok
test duplicate_slide_remaps_columns_children ... ok
test update_roundtrip ... ok
test move_slide_keeps_a_concurrent_text_edit ... ok
test remove_block_drops_it_from_columns_children ... ok
test move_block_keeps_a_concurrent_text_edit ... ok
test remaining_ops_roundtrip ... ok
test result: ok. 12 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s
Doc-tests slides_core
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
=== rust: fmt (pdf)
=== rust: clippy (pdf)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.15s
=== wasm: the pdf artifact the host embeds
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.06s
Finished `release` profile [optimized] target(s) in 0.13s
core/pdf/target/wasm32-unknown-unknown/release/slides_pdf.wasm (5301002 bytes)
=== the embedded pdf wasm must match a fresh build
=== rust: test (pdf)
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.07s
Running unittests src/lib.rs (core/pdf/target/debug/deps/slides_pdf-891fafaa44f9fd64)
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
Running tests/no_host_imports.rs (core/pdf/target/debug/deps/no_host_imports-17ccbb1d62120093)
running 1 test
test no_host_imports ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s
Running tests/render.rs (core/pdf/target/debug/deps/render-2e1d7e8b19d78595)
running 6 tests
test a_null_zero_length_pair_is_no_frames_not_a_dereference ... ok
test raster_image_is_embedded ... ok
test text_with_no_fonts_at_all_still_renders_the_shapes ... ok
test abi_frames_round_trip ... ok
test renders_text_as_embedded_glyphs ... ok
test two_svgs_make_two_pages ... ok
test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s
Doc-tests slides_pdf
running 0 tests
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
=== fixtures: node dependencies
v24.19.0
11.17.0
=== fixtures: npm ci (needs the registry — see .amadan/README.md)
added 19 packages, and audited 20 packages in 1s
5 packages are looking for funding
run `npm fund` for details
found 0 vulnerabilities
npm warn allow-scripts 1 package has install scripts not yet covered by allowScripts:
npm warn allow-scripts esbuild@0.28.2 (postinstall: node install.js)
npm warn allow-scripts
npm warn allow-scripts Run `npm approve-scripts --allow-scripts-pending` to review, or `npm approve-scripts <pkg>` to allow.
=== fixtures: the JS side's own tests
✔ one finished upload is register_asset plus the block that uses it, in one batch (30.483716ms)
✔ a record missing its numbers still makes an op the core accepts (1.2063ms)
✔ unusedAssets is what no slide refers to, and remove_asset drops one (5.490719ms)
✔ the Unused images list reconciles by id and names each Remove button (5.129379ms)
✔ an upload refusal is said in words the person can act on (0.284568ms)
✔ clampBox keeps the box inside the frame and no smaller than the minimum (1.765239ms)
✔ fixBox writes four decimals (0.256997ms)
✔ guides are the thirds plus every other block edge (0.305948ms)
✔ snapMove pulls the nearest edge onto a guide within one per cent, and leaves a far one (0.486384ms)
✔ snapResize moves only the edges the handle drags (0.296591ms)
✔ moveBox and resizeBox by handle (0.327399ms)
✔ nudge: arrows move by half a per cent, Shift by two, Alt resizes (0.281222ms)
✔ rectBox reads a block rectangle against its frame (0.250776ms)
✔ typing into a block is a text change on its slide only (37.908228ms)
✔ a block's props changing is structure on its slide (5.016148ms)
✔ adding, moving or removing a block is structure on its slide (18.88477ms)
✔ the notes are text on their slide (2.526602ms)
✔ a slide added, removed or moved is an order change (6.079063ms)
✔ one transaction touching several things reports each once (13.308341ms)
✔ what the ops write passes, and so does an empty document (10.036083ms)
✔ the shapes y-prosemirror and older documents write pass (1.879311ms)
✔ an unknown formatting attribute is reported with the core's message (2.22335ms)
✔ an element inside a paragraph, a non-paragraph child and bare text are reported (3.217668ms)
✔ the unknownmark poison shape is what the check refuses (1.854916ms)
✔ y-prosemirror 1.3.7 deletes the words under an unknown mark; sanitising first keeps them (5.35711ms)
✔ the vocabulary is the five marks and nothing else (0.159325ms)
✔ a reload reuses the id this tab stored for the deck (1.198547ms)
✔ two tabs, and two decks in one tab, get different ids (1.897134ms)
✔ a stored value that is not an id, or storage that throws, gives a fresh id (0.389824ms)
✔ a duplicated tab gives up the copied id; the original keeps it (12.385537ms)
✔ a reload finds no rival and keeps its id (4.086692ms)
✔ presence switches id: beats carry the new one and the old one is somebody else (3.053572ms)
✔ an empty body is no frames (1.583141ms)
✔ frames come back in order with their exact bytes (0.398119ms)
✔ a length prefix is big-endian (0.211203ms)
✔ a truncated frame throws rather than returning a short update (2.006478ms)
✔ deck controls are set_meta (14.978594ms)
✔ slide controls are set_slide, and background keeps its colour (4.249263ms)
✔ move up and down are move_slide with the right after (1.992612ms)
✔ block props: value kinds, region, a shape label (7.372404ms)
✔ box fields are set_block on the desktop and set_phone in the phone view (9.379634ms)
✔ move block and the layer buttons (2.471361ms)
✔ add block: region after the selection on flow, a default box on canvas, a heading level that fits (23.467202ms)
✔ a block is named by its type and its first words (1.996228ms)
✔ blockChange: a control element to its op (2.223881ms)
✔ the Blocks list reconciles by id and keeps its buttons (1.59351ms)
✔ open resolves to the button that closed it and the trimmed field (1.649204ms)
✔ a second open while the dialog is up joins the first (0.201154ms)
✔ a missing dialog mounts as nothing (0.13988ms)
✔ a tab advances its own cursor only when its POST is the very next row (2.895759ms)
✔ the cursor never moves backwards when a response is overtaken by this tab's own write (0.547106ms)
✔ a later response still carries the cursor forward (0.298074ms)
✔ a burst of notifications collapses into one fetch at a time (0.255014ms)
✔ an event naming a seq the tab already has fetches nothing (0.201775ms)
✔ a stopped subscriber drops a response that arrives late (0.278217ms)
✔ stop closes the stream, and start opens exactly one (0.235417ms)
✔ a hidden tab closes its stream on a shortened deadline, and activity reopens it (5.806976ms)
✔ switching away and back inside the grace never drops the stream, so nothing has to be caught up (0.349459ms)
✔ a visible tab that is only being READ closes when idle and comes back on any activity (1.467957ms)
✔ activity while streaming defers the idle close rather than reopening anything (0.153065ms)
✔ a stream failure is never reported as a save failure, and heals on activity (0.190814ms)
✔ an empty response advances the cursor without applying anything (0.840091ms)
✔ a failed fetch leaves the cursor alone so the next notification retries it (0.626995ms)
✔ a decode failure retries once through the raw log, then gives up (1.809141ms)
✔ updatesURL spells the query string exactly, in particular raw as the literal raw=1 (0.16706ms)
✔ empty_doc_views_as_defaults (26.603498ms)
✔ set_meta (14.091546ms)
✔ slides: add, move, remove, set_slide, set_notes (24.183053ms)
✔ duplicate_slide (14.018489ms)
✔ blocks: add, move, remove, set_block (19.562831ms)
✔ set_phone (6.169159ms)
✔ table cells follow rows and cols (14.099221ms)
✔ canvas blocks: box defaults from the region, z from the index (1.68504ms)
✔ assets: register and remove (2.218291ms)
✔ markdown: every mark, code blocks are plain lines (11.675768ms)
✔ batch_is_atomic (3.11188ms)
✔ a batch applies in one transaction with the given origin (1.003825ms)
✔ columns_children_validated (9.800975ms)
✔ layout_switch_round_trip (10.511094ms)
✔ view_matches_snapshot_fixtures (31.636939ms)
✔ columns children have one parent and no cycles (codex 7) (1.755952ms)
✔ canvas-to-flow keeps a concurrent text edit (codex 4) (5.154085ms)
✔ move_block keeps a concurrent text edit, and later inserts land where after says (codex 4) (4.100636ms)
✔ move_slide keeps a concurrent text edit, and later inserts land where after says (codex 4 at slide level) (4.409261ms)
✔ format: {name} and {n} interpolate; an unknown key is left as text (0.981203ms)
✔ a beat on connect, then one every twenty seconds (15.882422ms)
✔ a move posts once, 200ms after the last of a burst, and re-arms the beat (30.287111ms)
✔ a hidden tab stops beating; coming back posts at once (0.412146ms)
✔ applySet leaves this tab out and answers what changed (0.592601ms)
✔ a set with rubbish in it is read for what it does carry (0.267768ms)
✔ initials, and a client id per tab (32.968753ms)
✔ a resync that fails is retried with backoff until it succeeds, without ever running twice at once (14.148001ms)
✔ the browser's online event retries immediately, not just on the next backoff tick (0.448383ms)
✔ reportDivergence before any failure runs the first attempt without a caller ever calling online (1.0714ms)
✔ a permanent 4xx drops its batch instead of retrying it, and does not poison the next edit (85.404285ms)
✔ a network failure or a 5xx is retried, not dropped (60.27808ms)
✔ a 500 is retried the same way a network failure is, never dropped (66.645539ms)
✔ the core-written structure reads as ProseMirror marks (6.172035ms)
✔ a y-prosemirror round trip preserves bold and the link (2.306275ms)
✔ y-prosemirror writes href as the one-key map and marks as {} (0.713735ms)
✔ a plain-string href, written before the map shape, still reads (0.569729ms)
✔ textFragment writes the same shape as writeText (0.537548ms)
✔ safeHref is the server allowlist, and a disallowed href never draws an anchor (0.401205ms)
✔ parseHash splits the slide and the block, and survives a stray % (1.882808ms)
✔ destroy takes the listeners off, and a second mount leaves one of each (0.790349ms)
✔ formatHash is the inverse (8.818179ms)
✔ textBlocks finds the text-bearing blocks and skips the rest (3.748323ms)
✔ mountTarget is the element whose children are the paragraphs (0.327668ms)
✔ paragraphTag keeps the renderer's own markup per type (0.172ms)
✔ blockFragment finds a block's text by slide and block id (8.491082ms)
✔ notesFragment finds the slide's notes (1.554858ms)
✔ blockIDs and slideIDs are the orders on the stage and the strip (1.646258ms)
✔ cellFragments are a table's cells, row-major (1.742898ms)
✔ a load that fails leaves the stage on the slide it was showing and says so (1.80865ms)
✔ slideAfter is the slide to show once one is gone (0.541426ms)
✔ the queue words map to savestatus states, failed staying failed (0.987755ms)
✔ a dropped batch outranks the queue, for the rest of the session (0.183852ms)
✔ the editor status: the save line and the load error are two lines (0.255184ms)
✔ mountSaveStatus loads estilo's module and mounts it on the line (0.970072ms)
✔ a save-status module that will not load says so and refuses to mount (0.545293ms)
✔ dropTarget: before puts it ahead of the item, after behind it, null means first (0.99581ms)
✔ a drop that leaves the order as it is is no move (0.225519ms)
✔ dropOp is the move_slide (8.522031ms)
✔ labelFor is the slide's first heading, else the numbered fallback (8.217544ms)
✔ flush re-points every dirty thumbnail and relabels it (3.686158ms)
✔ destroy takes the item-menu listener off; a second mount leaves one (0.261335ms)
✔ plan says what to add, drop and where each item goes (1.136411ms)
✔ sync moves only the items out of place and keeps focus (1.542335ms)
✔ desktop view: box and size are the block; hidden and order are always the phone map (6.302718ms)
✔ phone view: everything goes to set_phone, carrying the rest of the map (0.633717ms)
✔ the ops it makes are ones the document takes (15.045167ms)
✔ phoneMap reads only the four keys, and only well-typed (0.257779ms)
✔ the stored view is read defensively (0.477167ms)
✔ linkAt reads the link under the cursor and nothing elsewhere (3.60183ms)
✔ setLink marks the selection (2.694075ms)
✔ setLink on a collapsed cursor inside a link replaces that link (1.107167ms)
✔ setLink with nothing selected and no link does nothing (0.314183ms)
✔ removeLink clears the link under the cursor (0.699039ms)
✔ the keymap binds every documented shortcut (0.259813ms)
✔ Mod-b toggles bold through the keymap (0.833268ms)
✔ setLink refuses an address outside the allowlist and writes nothing (0.788204ms)
✔ the state machine runs choose → sign → put → register → op (14.295584ms)
✔ a file the deck will not take never reaches the network (0.760853ms)
✔ alt text is required before the bytes move, unless the image is decorative (21.111957ms)
✔ a step that fails stops the ones after it and reports its own code (0.691355ms)
✔ a failure that is not ours still becomes an UploadError (0.353918ms)
✔ a second pick while one is in flight is refused, not raced (0.426342ms)
✔ an unreadable size is zero rather than a failure (0.293485ms)
✔ a per-call applyOp wins, so an upload lands on the block it was chosen for (0.309064ms)
✔ checkFile takes exactly the four types (0.478679ms)
✔ sha256Hex is the hex digest of the bytes (26.06647ms)
✔ refusal reads this app's JSON error shape, and survives one that is not (0.397067ms)
✔ malformed fonts are null (4.996863ms)
✔ a malformed background is the theme default (3.028285ms)
✔ a malformed box is null (3.522292ms)
✔ the phone layer keeps only its typed fields, order as a whole number (3.614294ms)
✔ assets are typed or skipped (0.623789ms)
✔ every poison update is deterministic bytes that apply to any deck (16.951059ms)
ℹ tests 158
ℹ suites 0
ℹ pass 158
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 697.594909
=== the vectors must match a fresh generation
canvas: 22 ops, 3715 bytes
columns: 23 ops, 3134 bytes
every-block: 36 ops, 3180 bytes
table: 14 ops, 1509 bytes
title-only: 6 ops, 1136 bytes
=== the editor bundle must match a fresh build
wrote internal/slides/static/editor.0463180baf5aab9f.js (376416 bytes)
=== go: vet, gofmt, test, migration check
go vet ./...
go test ./...
? amadan.net/oficina/slides/cmd/slides [no test files]
ok amadan.net/oficina/slides/internal/slides 13.608s
ok amadan.net/oficina/slides/internal/slides/core 3.048s
ok amadan.net/oficina/slides/internal/slides/deck (cached)
ok amadan.net/oficina/slides/internal/slides/fonts (cached)
? amadan.net/oficina/slides/internal/slides/icons [no test files]
? amadan.net/oficina/slides/internal/slides/keymailtest [no test files]
ok amadan.net/oficina/slides/internal/slides/layout (cached)
ok amadan.net/oficina/slides/internal/slides/live (cached)
ok amadan.net/oficina/slides/internal/slides/pdfcore (cached)
ok amadan.net/oficina/slides/internal/slides/present (cached)
ok amadan.net/oficina/slides/internal/slides/render (cached)
ok amadan.net/oficina/slides/internal/slides/svg (cached)
ok amadan.net/oficina/slides/internal/slidestest 42.276s
? amadan.net/oficina/slides/scripts/writeestilo [no test files]
go run amadan.net/rastrillo/rastrillo/cmd/rastrillo migration check
rastrillo migration check: models and migrations agree
python3 scripts/credits.py --check
credits: checked
gate: ok
diff --git a/internal/slides/app.go b/internal/slides/app.go| index 55f269c..af0cc97 100644 |
| --- a/internal/slides/app.go |
| +++ b/internal/slides/app.go |
| @@ -304,7 +304,7 @@ func newApp(d *db.DB, cfg Config) (*app, *rauth.Auth, error) { |
| // The sign-in mailer: links are queued and sent after the response, |
| // to permitted addresses only (mailqueue.go). Closed by app.Close. |
| - mailer := newMailQueue(cfg.Mailer, list.allows, logger) |
| + mailer := newMailQueue(cfg.Mailer, list.allows, nil, logger) |
| authPlugin, err := newAuthPlugin(cfg, d.Writer(), list, mailer) |
| if err != nil { |
| mailer.close() |
diff --git a/internal/slides/mailqueue.go b/internal/slides/mailqueue.go| index a0f9d55..27d227c 100644 |
| --- a/internal/slides/mailqueue.go |
| +++ b/internal/slides/mailqueue.go |
| @@ -2,111 +2,171 @@ package slides |
| import ( |
| "context" |
| + "fmt" |
| "log/slog" |
| "sync" |
| + "sync/atomic" |
| "time" |
| "amadan.net/rastrillo/rastrillo/mail" |
| ) |
| -// mailQueue is the sign-in mailer rauth is given: Send queues the link |
| -// and returns at once, and one background worker asks the allowlist and |
| -// only then talks to SMTP. |
| +// mailQueue is the sign-in mailer the auth plugin is given: Send queues |
| +// the link and returns at once, and one background worker decides |
| +// whether the address may have it and only then talks to SMTP. |
| // |
| -// Both halves are the no-oracle guarantee (gateBegin), measured rather |
| -// than reasoned about. When the send ran inside the request, a permitted |
| +// Both halves are the no-oracle guarantee, measured rather than |
| +// reasoned about. When the send ran inside the request, a permitted |
| // address waited for the SMTP round trip before its `303 ?sent=1` and a |
| -// denied one did not: 252 ms against 0.35 ms (Astra on #20). A response |
| -// time is an answer like any other. So POST /signin does the same |
| -// synchronous work for every address — rauth's budgets, its link row, |
| -// this enqueue — and the one admission-dependent step, the send, happens |
| -// after the response, where nobody can time it. The admission check |
| -// moved into the worker for the same reason: a check before responding |
| -// would be a branch an attacker could time, however cheap it is today. |
| +// denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides |
| +// #20). A response time is an answer like any other. So POST /signin |
| +// does the same synchronous work for every address — the auth plugin's |
| +// budgets, its link row, this enqueue — and the one admission-dependent |
| +// step, the send, happens after the response, where nobody can time it. |
| // |
| -// Delivery failures are logged, never surfaced. A failure only a |
| -// permitted address can reach is the oracle again, and the person is |
| -// told what every address is told: check your email, or ask again. |
| +// The request path is Send, and Send does no I/O and never blocks: not |
| +// a send, not an admission check, not even a log line. Every address's |
| +// link is queued, permitted or not, so how full the queue is never |
| +// depends on admission either; the worker decides afterwards. A full |
| +// queue drops the link and counts it in an atomic; the worker logs the |
| +// count when it next runs. The person is answered exactly as when the |
| +// queue is empty, because a different answer for a full queue would be |
| +// one more thing to read. (Astra on #20, round 2.) |
| // |
| -// Bounded: a queue of mailQueueDepth links. A full queue drops the link |
| -// with a log line instead of blocking a request — the per-IP and |
| -// per-address budgets inside Begin keep that from being an ordinary |
| -// event. Close drains what is queued, within mailDrainTimeout. |
| +// The worker survives a panic in the mailer: it recovers per message, |
| +// logs, and keeps draining. Delivery failures are logged, never |
| +// surfaced: a failure only a permitted address can reach is the oracle |
| +// again. |
| +// |
| +// close drains what is queued within drainTimeout. Past it, the send |
| +// in flight is cancelled through its context, everything still queued |
| +// is abandoned with a log line, and close waits a short grace for the |
| +// worker to stop. A mailer that ignores its context is the only way a |
| +// worker outlives close, and that is logged too. |
| type mailQueue struct { |
| next mail.Sender |
| - admits func(string) bool |
| + admits func(to string) bool |
| + // gate, when set, is read off the request's context at enqueue time |
| + // and travels with the link; the worker sends only if it said yes. |
| + gate func(context.Context) bool |
| logger *slog.Logger |
| - mu sync.Mutex |
| + drainTimeout time.Duration |
| + sendTimeout time.Duration |
| + |
| + mu sync.RWMutex // held for reading by Send, for writing by close |
| closed bool |
| jobs chan mailJob |
| + dropped atomic.Int64 |
| pending sync.WaitGroup |
| - done chan struct{} |
| + |
| + ctx context.Context // cancelled when close gives up waiting |
| + cancel context.CancelFunc |
| + done chan struct{} |
| } |
| -type mailJob struct{ to, subject, body string } |
| +type mailJob struct { |
| + to, subject, body string |
| + gated bool |
| +} |
| const ( |
| mailQueueDepth = 64 |
| mailSendTimeout = time.Minute |
| mailDrainTimeout = 30 * time.Second |
| + mailAbandonGrace = 5 * time.Second |
| ) |
| -func newMailQueue(next mail.Sender, admits func(string) bool, logger *slog.Logger) *mailQueue { |
| +func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger) *mailQueue { |
| if logger == nil { |
| logger = slog.Default() |
| } |
| - q := &mailQueue{next: next, admits: admits, logger: logger, |
| - jobs: make(chan mailJob, mailQueueDepth), done: make(chan struct{})} |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + q := &mailQueue{next: next, admits: admits, gate: gate, logger: logger, |
| + drainTimeout: mailDrainTimeout, sendTimeout: mailSendTimeout, |
| + jobs: make(chan mailJob, mailQueueDepth), ctx: ctx, cancel: cancel, |
| + done: make(chan struct{})} |
| go q.run() |
| return q |
| } |
| -// Send queues one link. It never blocks and never fails: the request |
| -// that called it answers the same way whatever happens next. |
| -func (q *mailQueue) Send(_ context.Context, to, subject, body string) error { |
| - q.mu.Lock() |
| - defer q.mu.Unlock() |
| +// Send queues one link. It never blocks, never does I/O and never |
| +// fails: the request that called it answers the same way whatever |
| +// happens next. |
| +func (q *mailQueue) Send(ctx context.Context, to, subject, body string) error { |
| + job := mailJob{to: to, subject: subject, body: body, gated: true} |
| + if q.gate != nil { |
| + job.gated = q.gate(ctx) |
| + } |
| + q.mu.RLock() |
| + defer q.mu.RUnlock() |
| if q.closed { |
| - q.logger.Error("slides: sign-in mail after shutdown; dropped") |
| + q.dropped.Add(1) |
| return nil |
| } |
| q.pending.Add(1) |
| select { |
| - case q.jobs <- mailJob{to: to, subject: subject, body: body}: |
| + case q.jobs <- job: |
| default: |
| q.pending.Done() |
| - q.logger.Error("slides: sign-in mail queue full; link dropped") |
| + q.dropped.Add(1) |
| } |
| return nil |
| } |
| func (q *mailQueue) run() { |
| defer close(q.done) |
| + abandoned := 0 |
| for job := range q.jobs { |
| - q.deliver(job) |
| + q.reportDrops() |
| + if q.ctx.Err() != nil { |
| + abandoned++ |
| + } else { |
| + q.deliver(job) |
| + } |
| q.pending.Done() |
| } |
| + q.reportDrops() |
| + if abandoned > 0 { |
| + q.logger.Error("slides: sign-in links abandoned at shutdown", "count", abandoned) |
| + } |
| } |
| +// reportDrops logs the links Send had to drop since the last report. |
| +// Here, on the worker, so the request path never writes a log line. |
| +func (q *mailQueue) reportDrops() { |
| + if n := q.dropped.Swap(0); n > 0 { |
| + q.logger.Error("slides: sign-in mail queue full; links dropped", "count", n) |
| + } |
| +} |
| + |
| +// deliver sends one link if it may go. A panic in the admission check |
| +// or the mailer is this message's failure, not the process's. |
| func (q *mailQueue) deliver(job mailJob) { |
| - if !q.admits(job.to) { |
| - // Not on the allowlist: the link rauth stored is never |
| + defer func() { |
| + if p := recover(); p != nil { |
| + q.logger.Error("slides: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p)) |
| + } |
| + }() |
| + if !job.gated || !q.admits(job.to) { |
| + // Not admitted: the link the auth plugin stored is never |
| // delivered, so it is never redeemed. No address in the log. |
| return |
| } |
| - ctx, cancel := context.WithTimeout(context.Background(), mailSendTimeout) |
| + ctx, cancel := context.WithTimeout(q.ctx, q.sendTimeout) |
| defer cancel() |
| if err := q.next.Send(ctx, job.to, job.subject, job.body); err != nil { |
| q.logger.Error("slides: send sign-in link", "err", err) |
| } |
| } |
| -// wait blocks until every link queued so far has been delivered or |
| -// dropped. Tests use it (App.FlushMail) to read the mailer afterwards. |
| +// wait blocks until every link queued so far has been sent or refused. |
| +// Tests use it to read the mailer afterwards. |
| func (q *mailQueue) wait() { q.pending.Wait() } |
| -// close stops taking links and drains what is queued. |
| +// close stops taking links and drains what is queued, within |
| +// drainTimeout; then it cancels, abandons the rest, and stops the |
| +// worker (see the type's comment). |
| func (q *mailQueue) close() { |
| q.mu.Lock() |
| if q.closed { |
| @@ -116,9 +176,16 @@ func (q *mailQueue) close() { |
| q.closed = true |
| close(q.jobs) |
| q.mu.Unlock() |
| + defer q.cancel() |
| + select { |
| + case <-q.done: |
| + return |
| + case <-time.After(q.drainTimeout): |
| + } |
| + q.cancel() |
| select { |
| case <-q.done: |
| - case <-time.After(mailDrainTimeout): |
| - q.logger.Error("slides: sign-in mail still sending at shutdown; abandoned") |
| + case <-time.After(mailAbandonGrace): |
| + q.logger.Error("slides: sign-in mail worker ignored cancellation; left running") |
| } |
| } |
diff --git a/internal/slides/mailqueue_test.go b/internal/slides/mailqueue_test.go| new file mode 100644 |
| index 0000000..8408ce9 |
| --- /dev/null |
| +++ b/internal/slides/mailqueue_test.go |
| @@ -0,0 +1,180 @@ |
| +package slides |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "log/slog" |
| + "runtime" |
| + "sync" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// stuckHandler is a slog handler whose every write blocks until release |
| +// is closed: a log sink that has stalled, as a full pipe does. |
| +type stuckHandler struct{ release chan struct{} } |
| + |
| +func (h stuckHandler) Enabled(context.Context, slog.Level) bool { return true } |
| +func (h stuckHandler) Handle(context.Context, slog.Record) error { |
| + <-h.release |
| + return nil |
| +} |
| +func (h stuckHandler) WithAttrs([]slog.Attr) slog.Handler { return h } |
| +func (h stuckHandler) WithGroup(string) slog.Handler { return h } |
| + |
| +// funcSender adapts a function to mail.Sender. |
| +type funcSender func(ctx context.Context, to string) error |
| + |
| +func (f funcSender) Send(ctx context.Context, to, _, _ string) error { return f(ctx, to) } |
| + |
| +func everyone(string) bool { return true } |
| + |
| +// TestMailQueueSendNeverBlocksWhenFull: Send is the request path, so it |
| +// may not block or do I/O even when the queue is full — not even to log |
| +// the drop (Astra on oficina/slides #20, round 2, P1: a "queue full" log |
| +// line on a stalled sink held the request, and saturation is |
| +// admission-dependent). The worker is stuck on its first link and the |
| +// log sink is stuck too; twice the queue's depth of Sends must still |
| +// return at once, and the drops are logged by the worker afterwards. |
| +func TestMailQueueSendNeverBlocksWhenFull(t *testing.T) { |
| + release := make(chan struct{}) |
| + var once sync.Once |
| + free := func() { once.Do(func() { close(release) }) } |
| + stuck := funcSender(func(ctx context.Context, _ string) error { |
| + select { |
| + case <-release: |
| + case <-ctx.Done(): |
| + } |
| + return nil |
| + }) |
| + q := newMailQueue(stuck, everyone, nil, slog.New(stuckHandler{release})) |
| + t.Cleanup(func() { free(); q.close() }) |
| + |
| + done := make(chan struct{}) |
| + go func() { |
| + for i := 0; i < 2*mailQueueDepth; i++ { |
| + _ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b") |
| + } |
| + close(done) |
| + }() |
| + select { |
| + case <-done: |
| + case <-time.After(2 * time.Second): |
| + free() |
| + <-done |
| + t.Fatal("Send blocked with the queue full: the request path waited on the queue or the log") |
| + } |
| + if n := q.dropped.Load(); n == 0 { |
| + t.Fatal("no drop was counted with twice the queue's depth sent") |
| + } |
| +} |
| + |
| +// TestMailQueueQueuesEveryAddress: the queue's fill level must not |
| +// depend on admission, so Send queues a refused address's link exactly |
| +// as it queues a permitted one; only the worker asks. |
| +func TestMailQueueQueuesEveryAddress(t *testing.T) { |
| + release := make(chan struct{}) |
| + stuck := funcSender(func(ctx context.Context, _ string) error { |
| + select { |
| + case <-release: |
| + case <-ctx.Done(): |
| + } |
| + return nil |
| + }) |
| + var asked sync.Map |
| + admits := func(to string) bool { asked.Store(to, true); return to == "first@example.test" } |
| + q := newMailQueue(stuck, admits, nil, slog.New(slog.DiscardHandler)) |
| + t.Cleanup(func() { close(release); q.close() }) |
| + _ = q.Send(context.Background(), "first@example.test", "s", "b") |
| + deadline := time.Now().Add(2 * time.Second) |
| + for _, ok := asked.Load("first@example.test"); !ok && time.Now().Before(deadline); _, ok = asked.Load("first@example.test") { |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + for i := 0; i < 5; i++ { |
| + _ = q.Send(context.Background(), fmt.Sprintf("refused%d@example.test", i), "s", "b") |
| + } |
| + if got := len(q.jobs); got != 5 { |
| + t.Fatalf("queued %d refused links, want all 5: fill level must not depend on admission", got) |
| + } |
| +} |
| + |
| +// TestMailQueueSurvivesAPanic: a panic in the mailer is that message's |
| +// failure; the worker recovers, logs, and sends the next one (Astra, |
| +// round 2, P2). |
| +func TestMailQueueSurvivesAPanic(t *testing.T) { |
| + var mu sync.Mutex |
| + var sent []string |
| + calls := 0 |
| + s := funcSender(func(_ context.Context, to string) error { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + calls++ |
| + if calls == 1 { |
| + panic("mailer exploded") |
| + } |
| + sent = append(sent, to) |
| + return nil |
| + }) |
| + q := newMailQueue(s, everyone, nil, slog.New(slog.DiscardHandler)) |
| + t.Cleanup(q.close) |
| + _ = q.Send(context.Background(), "first@example.test", "s", "b") |
| + _ = q.Send(context.Background(), "second@example.test", "s", "b") |
| + q.wait() |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if len(sent) != 1 || sent[0] != "second@example.test" { |
| + t.Fatalf("sent = %v, want the second link after the first panicked", sent) |
| + } |
| +} |
| + |
| +// TestMailQueueCloseIsBoundedAndStopsTheWorker: close drains within its |
| +// deadline, then cancels the send in flight, abandons the rest, and the |
| +// worker is gone — no goroutine outlives close (Astra, round 2, P2). |
| +func TestMailQueueCloseIsBoundedAndStopsTheWorker(t *testing.T) { |
| + baseline := runtime.NumGoroutine() |
| + honours := funcSender(func(ctx context.Context, _ string) error { |
| + <-ctx.Done() // a relay that never answers, but obeys cancellation |
| + return ctx.Err() |
| + }) |
| + q := newMailQueue(honours, everyone, nil, slog.New(slog.DiscardHandler)) |
| + q.drainTimeout = 50 * time.Millisecond |
| + for i := 0; i < 3; i++ { |
| + _ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b") |
| + } |
| + start := time.Now() |
| + q.close() |
| + if took := time.Since(start); took > 2*time.Second { |
| + t.Fatalf("close took %v, want about its 50ms deadline", took) |
| + } |
| + select { |
| + case <-q.done: |
| + default: |
| + t.Fatal("the worker was still running after close") |
| + } |
| + deadline := time.Now().Add(time.Second) |
| + for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + if n := runtime.NumGoroutine(); n > baseline { |
| + t.Fatalf("%d goroutines after close, %d before: the queue leaked", n, baseline) |
| + } |
| +} |
| + |
| +// TestAppCloseStopsTheMailWorker: the app owns its queue, and Close |
| +// stops its worker — no test that builds an app and closes it leaves |
| +// one running behind it. |
| +func TestAppCloseStopsTheMailWorker(t *testing.T) { |
| + _, a, err := New(testDB(t), Config{Env: EnvDevelopment, Origin: "http://localhost:8091", Logger: quietLogger()}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + q := a.mail |
| + if err := a.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + select { |
| + case <-q.done: |
| + case <-time.After(2 * time.Second): |
| + t.Fatal("the sign-in mail worker was still running after app.Close") |
| + } |
| +} |
diff --git a/internal/slidestest/signin_timing_test.go b/internal/slidestest/signin_timing_test.go| index 9fd99b0..461f884 100644 |
| --- a/internal/slidestest/signin_timing_test.go |
| +++ b/internal/slidestest/signin_timing_test.go |
| @@ -2,8 +2,11 @@ package slidestest |
| import ( |
| "context" |
| + "fmt" |
| + "net/http" |
| "net/http/httptest" |
| "net/url" |
| + "strings" |
| "sync" |
| "testing" |
| "time" |
| @@ -81,3 +84,42 @@ func TestSigninNeverWaitsForTheMailer(t *testing.T) { |
| t.Fatalf("mail sent = %v, want two links, both to %s", got, allowedAddress) |
| } |
| } |
| + |
| +// TestAFullMailQueueAnswersLikeAnEmptyOne: the queue holds 64 links. |
| +// With the mailer stuck on the first (permitted) link, enough POSTs to |
| +// overflow it — each from its own address and IP, so no budget stands |
| +// in — must every one be answered exactly as the first was, and at |
| +// once. Every address's link is queued, permitted or not, so a refused |
| +// address fills the queue as surely as a permitted one (Astra on #20, |
| +// round 2). |
| +func TestAFullMailQueueAnswersLikeAnEmptyOne(t *testing.T) { |
| + bm := &blockingMailer{release: make(chan struct{})} |
| + cfg, _ := newConfig(t) |
| + cfg.Mailer = bm |
| + app := newAppWith(t, cfg, nil) |
| + var once sync.Once |
| + free := func() { once.Do(func() { close(bm.release) }) } |
| + t.Cleanup(free) |
| + |
| + postFrom := func(i int, address string) *httptest.ResponseRecorder { |
| + req := httptest.NewRequest(http.MethodPost, "/signin", strings.NewReader(url.Values{"address": {address}}.Encode())) |
| + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") |
| + req.Header.Set("Origin", testOrigin) |
| + req.Header.Set("Sec-Fetch-Site", "same-origin") |
| + req.RemoteAddr = fmt.Sprintf("198.51.100.%d:1234", i%250+1) |
| + rec := httptest.NewRecorder() |
| + app.ServeHTTP(rec, req) |
| + return rec |
| + } |
| + first := postFrom(0, allowedAddress) |
| + for i := 1; i <= 80; i++ { |
| + start := time.Now() |
| + rec := postFrom(i, fmt.Sprintf("stranger%d@else.test", i)) |
| + if took := time.Since(start); took > time.Second { |
| + t.Fatalf("POST %d took %v with the queue full", i, took) |
| + } |
| + if rec.Code != first.Code || rec.Header().Get("Location") != first.Header().Get("Location") || rec.Body.String() != first.Body.String() { |
| + t.Fatalf("POST %d answered %d %q, want exactly the first answer %d %q", i, rec.Code, rec.Header().Get("Location"), first.Code, first.Header().Get("Location")) |
| + } |
| + } |
| +} |