Repository navigation
feat: store received mail (inbound Phase B) - #61
Merged
Merged
Conversation
Otitodev
changed the title
Two names settled before the code that uses them, so the model, the service and the worker do not each coin their own. The prefix is "inbound", so a received message reads inbound_01J... in the API. Like every prefix it is permanent once shipped. INBOUND_RETENTION_DAYS (default 30) is how long the raw message and its attachments stay in the project's bucket. The parsed message stays in the database either way, so this bounds only what costs storage and holds somebody's files. It is bounded to 1..3650: zero would have S3 expire mail the moment it lands, and no upper bound turns a typo into "never". Refused at startup, where the error names the setting, rather than at provisioning where AWS would say something about a lifecycle rule. A change applies the next time receiving is set up; an object already stored keeps the clock it had. Polling reuses the EVENT_POLL_* settings rather than adding a second set that would have to be kept in step with them.
The inverse of message.py, and a different problem. That module builds mail SESKit wrote, so a declared charset is a real one and a multipart is well formed. This reads mail somebody else wrote, and none of that holds. It never raises. A message that cannot be read is still a message somebody sent, and a worker that raises on it leaves it on the queue to be retried for ever. The worst case is parse_failed set on a result that is stored and shown. Nothing leaves it that the database would refuse. Postgres cannot store a NUL in text and the driver cannot encode a lone surrogate, which 8-bit header bytes produce; either turns one odd message into an insert that fails on every retry. _clean is the one place that removes them, and the tests were run with it disabled to confirm they fail without it. Bounded on every axis - parts, nesting depth, headers, body length - and a cap sets `truncated` rather than failing, because a 40 MB message is legitimate and a 40 MB message of nothing but nested multiparts is an attack that looks identical until something counts. A nesting bomb three thousand deep returns a result. Attachments are described, not carried. The bytes stay in the stored message, where they already are; read_attachment finds one by index by walking the same parts in the same order, so an index means the same thing in both places. A message/rfc822 part is one attachment: descending into it would take a forwarded message's body for this one's. Filenames are reduced to a basename with control characters removed, because the sender chose the string and a download will echo it in a header. A backslash path inside a plain quoted filename is consumed as an escape by RFC 2045 before this module sees it, which looks like a bug and is the reason it is safe; a test pins that. The real route for one is RFC 2231 percent-encoding, which is tested. Only the first text/plain and first text/html are bodies. Any further text part stays visible as an attachment: nothing is silently dropped.
The table the messages go in, the columns that record the plumbing built to bring them, and the column that records which domain's mail arrives through it. Nothing reads or writes them yet. inbound_emails mirrors Email in name only. Email records something SESKit was asked to do, written before it happens; this records something that already happened to somebody else's message. The parsed message is stored whole and stays as long as the row does; the original stays in the bucket until retention removes it, and raw_expires_at says when. So a message can still be read after its attachments can no longer be downloaded, which is the right way round - the row costs almost nothing and the files cost storage. Two uniqueness rules, because each catches what the other cannot. SNS is at-least-once, so the same notification arrives twice and its message id says so. SES can also announce one stored object again under a different SNS id, which only the storage location notices. A message holds its domain as text and not a reference to the identity: removing a domain stops future mail and must not take what was already received with it. One receiving identity per domain is enforced by a partial unique index on identities.value, not left to the service layer. A domain's MX record names a single endpoint in a single region, so two identities both receiving for it cannot both be right, and mail would belong to whichever was looked up first. Partial because most identities receive nothing and any number may share a value; a check-then-insert in a service cannot make that promise under two concurrent requests, an index can. The rule name and rule set are recorded on the identity rather than derived, so teardown removes the rule that was created, from the set it was put in. That is whichever set the account had active at the time and no name can reconstruct it. inbound_rule_set_created records whether SESKit made the set, because only then may it remove it. The connection keeps inbound_bucket after the topic and queue are gone if the bucket still holds mail, so a later setup finds it again. inbound_enabled therefore asks for the topic and queue, not the bucket: a kept bucket does not mean anything is listening. The references column is reference_ids: references is reserved in Postgres and a column that must be quoted in every hand-written query is a trap. The migration is written by hand, since autogenerate needs a database. Checked offline instead: alembic emits the expected SQL in both directions, there is one head, and the model DDL compiles. The check that runs it against a real database and diffs it against the models is tests/test_migrations.py, in CI.
The service layer over the provider: decide whether to create or remove anything, and record what happened. Nothing calls setup yet - the routes that turn it on are a later phase - but removal is wired in now, because that is the half that must never be missed. Two halves, owned differently. The bucket, topic and queue belong to an AWS account and region and are recorded on each project's connection; a receipt rule belongs to one domain and is recorded on its identity. A second domain adds its rule and leaves the first's plumbing alone. The refcount is the reason for most of this file, and the same trap services.events documents. Two projects in one account and region share the plumbing, so removing it when the first project's last domain goes stops the second project's mail, with nothing on either screen to say why. Teardown counts the other users first, matched on account and region rather than on a stored ARN. A bucket that still holds mail stays, and stays recorded, so a later setup finds it and the name is not lost with the row that knew it. Every removal touches AWS first and the row second. The other way round, a failure halfway leaves a row saying "not receiving" while the rule is still in somebody's account and nothing left to name it. A test makes the removal fail and asserts the row still says it receives. The domain is claimed before anything is built. The name is flushed first, inside a savepoint, so the unique index decides a race between two requests enabling the same domain: the loser fails before it has created a rule in somebody's account, and does not roll back the request that lost. The friendly pre-check beside it is only courtesy; a check followed by an insert cannot hold under concurrency, an index can. The refusal does not say whose domain it is. remove_identity and disconnect_aws now take an optional inbound factory and raise ValueError when something receives and none is given. Deleting the row would strand the rule, since nothing could then say which receipt rule in the account was ours; disconnecting would drop the only key that could remove it. A project that receives nothing, which is every project today, needs no factory and behaves as before. The two routes that call them pass one. Core cannot import the adapter, so the rule-name prefix is repeated here with a test that holds it equal to the adapter's. A mismatch would let SES accept a rule whose writes the bucket policy then refuses.
The counterpart of ingest_event, for mail that arrives instead of mail that was sent. Nothing calls it yet; the worker job is next. Split into steps because the worker has to do something between them that this module should not know about: fetch the message. is_recorded is its own step so a redelivery of a 40 MB message costs a database lookup and not a download. Whose message it is comes from the recipient's domain, not the object key. SES says the key equals the message id and is silent on whether a rule's key prefix is included, so nothing here parses it; the envelope recipients are what the rule matched. A rule for example.com also matches sub.example.com, so a recipient's parent domains are candidates too, most specific first. That match is the lock, for the reason ingest_event's message-id scope is. Nothing in a notification is secret - the SES message id is in the headers of every message it delivers - so an announcement naming a domain that belongs to a project the queue does not speak for, or a bucket that is not this queue's, is dropped. Every way of being wrong returns the same None, so a probe learns nothing from the difference. An empty scope matches nothing. A verdict the notification omits is stored as NULL, never as PASS. Saying a message was authenticated when nothing checked it is the one thing this column must not do. Verdicts are recorded and not acted on. Losing a race to record the same notification ends as a duplicate and not an error, inside a savepoint, so the loser does not roll back whatever the caller has pending; a test records another message straight afterwards in the same session to prove it. The parser's cleaning is proved here, where it can be: real NUL bytes and invalid UTF-8 going into real Postgres, which is what refuses them. And a message nobody can read is still recorded, with the original in storage, instead of left on the queue to fail for ever. Logs carry ids and sizes only. Who a message is from, who it is to and what it says are not a reason to put any of it in a log.
poll_inbound drains each inbound queue, fetches the message an announcement points at, and records it. It runs every minute, offset from the event poll so the two long polls do not start together. The last piece of storing received mail: nothing receives yet, since nothing turns receiving on, but everything below that line now works end to end. The order puts the expensive step last. Whose message it is and whether it is already stored are database lookups; the download is up to 40 MB. A redelivery, a message for somebody else's domain, and a notification that is not mail at all never reach it, and tests assert the store was not asked. The download holds no database connection, so the lookups, the fetch and the write are separate short sessions. What acknowledges a message follows events.py: deleted only once settled - recorded, already recorded, not ours, not mail, or expired. An expired object is settled, because it will not come back and a queue that keeps redelivering it makes no progress on the rest. A download that failed for a reason that may pass - a permission being added, a network returning - is left on the queue, because deleting it would turn a fixable setup mistake into mail that is gone. The one deliberate difference from events.py is isolation. There an unexpected exception ends the pass; here the step that can fail is a download, and the commonest cause is a missing s3:GetObject, which fails identically on every message. If one failure ended the batch, the unreadable message at the front would stall everything behind it indefinitely. Each message is isolated instead: the failure is logged by type and the queue's own id - never its text, since an exception raised while handling mail can carry a fragment of it - the message is left for the next pass, and the rest of the batch is handled. A test fails the first of three downloads and asserts the other two are recorded. The original is never deleted from the bucket here; retention does that. A bounded pass and a single request on an empty queue, as before.
Otitodev
force-pushed
the
feat/inbound-storage
branch
from
October 5, 2026 04:45
b8c98ae to
a453cec
Compare
This was referenced Oct 5, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What and why
Follows #60, which is merged; this is rebased onto
main.SESKit can send mail and report what happened to it, but cannot receive. #60 added the SES half. This adds everything that happens once a message has arrived: storing it, parsing it, and a worker job that reads the queue. Nothing turns receiving on yet, since that is the routes and dashboard in a later PR, so no user-visible behaviour changes. What is new is the path from "SES stored a message" to "a parsed row in the database".
Six commits, one concern each:
inbound_01J....INBOUND_RETENTION_DAYS(default 30, bounded 1 to 3650) is how long the raw message and attachments stay in S3; the parsed message stays in the database regardless.inbound_emailstable, the columns that record the plumbing, and a partial unique index so one domain can receive in only one place.poll_inbound, every minute.What to review hardest:
test_plumbing_stays_while_another_project_in_the_account_still_receivesis the guard.s3:GetObjectfails identically on every message.remove_identityanddisconnect_awsgained an optional factory and raiseValueErrorwhen something receives and none is given. Every project today receives nothing and behaves as before; the two routes that call them pass one.The 22 service tests, 19 ingest tests and 12 model tests run against Postgres and ran in CI. Locally I could only run the static gate and the tests that need no database, since I did not start Docker.
Not yet verified against a real AWS account: that SES accepts a rule publishing to our topic without a topic policy (moto does), and whether a rule's key prefix appears in the reported
objectKey. The ingest code deliberately does not depend on the second.Checks
tests/test_migrations.pypasses in CI. I generated the SQL offline and checked it; I did not runalembic upgrade headagainst a database myself.uv run --group docs mkdocs build --strictis clean🤖 Generated with Claude Code