feat(plumber): retention policy worker #717
Conversation
skipi
force-pushed
the
mk/plumber/retention-policies
branch
4 times, most recently
from
7e03b42 to
3b83c30
Compare
skipi
force-pushed
the
mk/plumber/retention-policies
branch
from
dd35670 to
18191c5
Compare
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
plumber/ppl/priv/ecto_repo/migrations/20251201141215_add_expires_at_to_pipeline_requests.exs
Show resolved
Hide resolved
skipi
force-pushed
the
mk/plumber/retention-policies
branch
from
aaac102 to
af9fec5
Compare
| expired_records = | ||
| from(pr in PplRequests, | ||
| where: pr.expires_at < ^now, | ||
| select: %{id: pr.id, ppl_artefact_id: pr.ppl_artefact_id}, | ||
| limit: ^limit | ||
| ) | ||
| |> EctoRepo.all() |
There was a problem hiding this comment.
Just a comment, but try to run this with the explain to validate that the planner picks the index that you created to run this query. (It should infer expires_at is not null but it may not)
There was a problem hiding this comment.
I've added additional index for this:
|
|
||
| if days < @min_grace_period_days do | ||
| Logger.warning( | ||
| "[Retention] grace_period_days=#{days} is below minimum, using #{@min_grace_period_days}" |
There was a problem hiding this comment.
Let's add org_id in this log line, so it is easier to debug.
| ) | ||
|
|
||
| {marked_count, _} = EctoRepo.update_all(mark_query, set: [expires_at: expires_at]) | ||
| {unmarked_count, _} = EctoRepo.update_all(unmark_query, set: [expires_at: nil]) |
There was a problem hiding this comment.
These update_all calls can cause issues since if the cutoff date is recent enough, we might end up updating a lot of rows at once and have issues with locks, heavy IO, connection timeout, etc.
The more secure way here would be to do batching with a limit on the number of rows that are updated at once.
skipi
force-pushed
the
mk/plumber/retention-policies
branch
from
af9fec5 to
0d49ae0
Compare
📝 Description
This PR implements retention policy support in plumber. When the usage service applies a retention policy, plumber marks pipeline records for expiration with a configurable grace period, allowing corrections before data is deleted.
expires_atcolumn topipeline_requeststable to track when records should be deletedPolicyConsumerto receive retention policy events from usage service via RabbitMQPolicyApplierto mark/unmark pipeline records based on the cutoff dateexpires_at = now + grace_period(default 15 days)expires_atcleared, allowing policy corrections to "save" mistakenly marked recordsRETENTION_GRACE_PERIOD_DAYSenv var (min: 7 days)✅ Checklist