Feat(core): [1083] transformation system packer by mmascher · Pull Request #1102 · DIRACGrid/diracx

mmascher · 2026-10-21T08:37:55Z

Summary

Implements the Transformation System packer (#1083, part of #1081), following DX-ADR-006 §Packer. The packer is the core task that turns a transformation's Unassigned inputs into parcels. It also includes one simple packer plugin, tests and developer docs.

The rule from the ADR runs through the design: plugins propose, the core writes. A plugin receives a stream of claimed inputs and yields proposals; only the core touches the database.

What's included

The commits follow package boundaries and can be reviewed in order:

core: the plugin contract models in diracx.core.models.transformation: PackerInput, plus the proposals ComputeParcel, DataParcel, BadInput and DelayInput. Also the new diracx.transformations.packers entry-point group and two exceptions.
db: bulk TransformationDB methods with no domain logic:
- claiming Unassigned inputs with FOR UPDATE SKIP LOCKED and keyset paging
- inserting compute parcels and assigning their inputs
- marking inputs Problematic or delayed
- Every status change is guarded with WHERE Status = 'Unassigned' plus a rowcount check.
logic: pack_transformation, which:
- streams claimed inputs to the plugin
- checks each proposal (the input was offered, and is not used twice)
- merges requirements over the transformation's template and stores them by content hash
- writes in batches
Also FixedSize, a packer that makes parcels of group_size inputs and holds back a smaller remainder until the feeder is off or a flush is requested.
tasks: PackerSweepTask, which runs periodically and schedules one PackTransformationTask per Active transformation. Each task holds a mutex on its transformation. The sweep is disabled by default (DIRACX_TASKS_PACKER_ENABLED).
docs: a developer page on the contract and how to write and register a packer, plus the new settings in the env-variable reference.

Scope decisions

Compute only. DataParcel is defined; the core raises NotImplementedError on it and on Data transformations.
No counters yet. Every place that should write a journal row has a TODO(#1085). The input overview uses a GROUP BY until then.
No dependency injection into plugins. They receive only the contract arguments; the injection DX-ADR-006 describes is a follow-up.
Tasks are registered under the diracx.tasks.transformation group.
Inputs has no reason column, so the reason for a BadInput goes to TransformationLog.

Testing

Unit tests for the db, logic and tasks layers, all on SQLite.
Tried end to end with pixi run local-start: a seeded transformation with 10 inputs and group_size=3 became 4 parcels.
Not yet run against MySQL, which matters because SQLite ignores SKIP LOCKED.

Open points

Worker commits the injected DB transaction when a task fails #1101: the worker commits an injected transaction even when a task fails. So until that's fixed, a packer run that fails part-way keeps its earlier writes. It will be fixed in the worker, separately from this PR.
Lock key shared with the feeder ([Transformation System] Feeder #1082): under discussion in the review thread on packer.py.
Raised in review, still to discuss:
- The status is read without locking the Transformations row.
- Performance of the LastParcelID executemany for about 10k inputs.
- DelayInput is applied immediately rather than batched.
- group_size larger than max_inputs.

This stays a draft until these points are settled.

Add the proposal models a packer plugin yields (ComputeParcel, DataParcel, BadInput, DelayInput), the PackerInput it receives, the diracx.transformations.packers entry-point group, and the TransformationNotFoundError and InputStatusConflictError exceptions. Refs DIRACGrid#1083 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Add the bulk operations the packer needs: reading a transformation and its compute facet, counting inputs by status, claiming Unassigned inputs with FOR UPDATE SKIP LOCKED, inserting compute parcels and assigning their inputs, marking inputs Problematic and delaying them. Counter journalling is left to DIRACGrid#1085. Refs DIRACGrid#1083 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

pack_transformation claims a transformation's Unassigned inputs, streams them to the packer plugin named by the transformation, checks every proposal (offered, not reused) and applies it: compute parcels with their requirements merged over the template, Problematic inputs and delays. Data parcels and dependency injection into plugins are not supported yet. FixedSize is a simple packer making parcels of a fixed number of inputs, holding back the remainder until the feeder is off or on a flush. Refs DIRACGrid#1083 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…on task PackerSweepTask periodically schedules a PackTransformationTask for every Active transformation; each one holds a mutex on its transformation and calls pack_transformation. The sweep is disabled by default and configured through DIRACX_TASKS_PACKER_*. Refs DIRACGrid#1083 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Add a developer page on the packer contract, what the core checks and how to register a packer through the diracx.transformations.packers entry point, and document the DIRACX_TASKS_PACKER_* settings. Refs DIRACGrid#1083 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

The task worker commits an injected DB transaction even when the task raises, so a packer run that fails part-way keeps its earlier writes. Point to the issue where the packer task and pack_transformation claim a rollback. Refs DIRACGrid#1083, DIRACGrid#1101 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…cker

mmascher · 2026-10-21T08:38:55Z

This is still in draft, there are a few things I'd like to check, including #1101

read-the-docs-community · 2026-10-21T08:39:29Z

Documentation build overview

📚 diracx | 🛠️ Build #35322958 | 📁 Comparing b57e364 against latest (0452593)

🔍 Preview build

2 files changed

+ dev/explanations/transformation-system/packer/index.html
± admin/reference/env-variables/index.html

Copilot

🟡 Changes recommended

Schema mismatches break imports and requirements persistence, while accepted grouping settings can prevent packing progress indefinitely.

3 open issues

What changed in this PR

Adds plugin-based packing to the Transformation System, turning unassigned inputs into compute parcels through core-controlled database writes.

Changes:

Defines packer contracts and a built-in fixed-size plugin.
Adds packing logic, database operations and scheduled tasks.
Adds tests and configuration documentation.

File	Description
mkdocs.yml	Adds packer documentation navigation.
docs/dev/explanations/transformation-system/packer.md	Explains packing and plugin development.
docs/admin/reference/env-variables.md.j2	Includes packer settings generation.
docs/admin/reference/env-variables.md	Documents packer environment variables.
diracx-tasks/tests/test_packer_tasks.py	Tests task registration and delegation.
diracx-tasks/src/diracx/tasks/transformations/packer.py	Adds sweep and per-transformation tasks.
diracx-tasks/src/diracx/tasks/transformations/__init__.py	Exports packing tasks.
diracx-tasks/pyproject.toml	Registers task entry points.
diracx-logic/tests/transformations/test_packing.py	Tests packing orchestration and persistence.
diracx-logic/tests/transformations/test_packers.py	Tests fixed-size grouping.
diracx-logic/src/diracx/logic/transformations/packing.py	Loads plugins and applies validated proposals.
diracx-logic/src/diracx/logic/transformations/packers.py	Implements fixed-size packing.
diracx-logic/src/diracx/logic/transformations/__init__.py	Exports packing interfaces.
diracx-logic/pyproject.toml	Registers the fixed-size plugin.
diracx-db/tests/transformation/test_transformation_db.py	Tests packing database operations.
diracx-db/src/diracx/db/sql/transformation/db.py	Adds input claiming and parcel persistence.
diracx-core/tests/test_transformation_models.py	Tests packer model validation.
diracx-core/src/diracx/core/models/transformation.py	Defines packer inputs and proposals.
diracx-core/src/diracx/core/models/__init__.py	Exports packer models.
diracx-core/src/diracx/core/extensions.py	Defines the packer entry-point group.
diracx-core/src/diracx/core/exceptions.py	Adds transformation database exceptions.

_{Review effort: Balanced}

mmascher · 2026-10-21T14:06:06Z

+    Inputs,
+    ParcelInputs,
+    Parcels,
+    Requirements,


Makes sense, a column rename happened after I developed the code.

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

francescobrivio

Just reporting a few comments from a Claude Artifact review of this code that we (me and Marco) discussed yesterday and we believe should probably be discussed with the experts.

francescobrivio · 2026-10-21T14:11:02Z

+
+    @property
+    def execution_locks(self) -> list[BaseLock]:
+        return [MutexLock(TRANSFORMATION, self.transformation_id)]


Each transformation will have to be also "acquired" by the feeder task (still being developed) which will be started exactly at the same time of the packer (when the Transformation goes from NEW to ACTIVE) and executed as a PeriodicTask.

Will the MutexLock prevent this? If so, could we simply specify more the lock with:

Suggested change

return [MutexLock(TRANSFORMATION, self.transformation_id)]

return [MutexLock(TRANSFORMATION, self.transformation_id, "packer")]

Would this be enough? Or do we need to use a different Lock?

I am learning the ropes of the project, but why you say that Feeder and Packer would start exactly at the same time? I expected it to be asynchronous and based on the database state (so, potentially running right after the transaction is comited, but not necessarily).

Here I'm talking about the "Feeder and Packer DiracX-Tasks" which periodically look at transformations and when a transformation goes from NEW to ACTIVE they "start acting on it", so they can potentially detect the state change of a transformation at the same time and they potentially want to lock it at the same time.

At least this is what I understood...I might be completely wrong on how the locking works 😄

I think having a packer/feeder specific lock makes sense. There is no reason why they can't run in parallel to each other. We then need a reader-writer lock to ensure the status is respected correctly. i.e.

Feeder: MutexLock(TRANSFORMATION, id, "feeder") plus SharedRWLock(TRANSFORMATION, id, "status")

Packer: MutexLock(TRANSFORMATION, id, "packer") plus SharedRWLock(TRANSFORMATION, id, "status")

Anything that changes status (hooks, drain/finalize sweeps, actions): ExclusiveRWLock(TRANSFORMATION, id, "status")

So both locks applied at the same time right?

Is there an example somewhere in dirac/diracx how that should be coded technically?

francescobrivio · 2026-10-21T14:17:57Z

+        # Each input points at its own parcel, so this one is an executemany
+        await self.conn.execute(
+            update(Inputs)
+            .where(
+                Inputs.transformation_id == bindparam("b_transformation_id"),
+                Inputs.input_id == bindparam("b_input_id"),
+            )
+            .values(last_parcel_id=bindparam("b_parcel_id")),
+            [
+                {
+                    "b_transformation_id": transformation_id,
+                    "b_input_id": link["InputID"],
+                    "b_parcel_id": link["ParcelID"],
+                }
+                for link in links
+            ],
+        )


The Claude artifact we used to review this PR had a comment on this lines saying that this is an executemany (see also the comment line) asking if from a performance on aiomysql point of view it would be acceptable for 10k inputs?

(Not sure I have a smart comment about this, just reporting for the sake of clarity/review)

francescobrivio · 2026-10-21T14:19:32Z

+    if transformation["status"] != TransformationStatus.ACTIVE:
+        return PackingSummary(skipped=True)
+    if transformation["kind"] != TransformationKind.COMPUTE:
+        raise NotImplementedError("Packing data transformations is not supported yet")


the same Claude artifact points out here:

The status is read without locking the Transformations row, so a concurrent Pause does not stop a run that has already started.

To be discussed I guess?

francescobrivio · 2026-10-21T14:20:29Z

+                await transformation_db.delay_inputs(
+                    transformation_id, proposal.input_ids, proposal.delay_until
+                )


and again, from the same artifact:

Applied straight away rather than buffered like parcels; one statement per DelayInput.

chrisburr · 2026-10-22T08:36:13Z

+
+
+def test_packer_input_is_frozen():
+    packer_input = PackerInput(input_id=uuid4(), lfn="/vo/file")


It's important to use uuid v7 and not v4, see #486

The merge with main brought in the rename of Requirements to RequirementsTemplates, which TransformationDB and the packer tests still imported under the old name, so importing diracx.db.sql failed. Refs DIRACGrid#1083

amaltaro · 2026-10-22T12:46:46Z

My understanding is that this PR is supposed to give the abstractions for DiracX packer functionality.
If that is correct, then I think it could come with the most simplistic packer implementation, e.g.: something that generates one parcel for each input, leaving configurations of input pool size, grouping size and output pool size for future implementations.

amaltaro · 2026-10-22T14:46:12Z

+    inputs: list[UUID] = Field(min_length=1)
+    metadata: dict[str, Any] = {}
+    requirements: dict[str, Any] | None = None
+    parameters: dict[str, Any] = {}


If this is the CWL compute transformation (the step itself), shouldn't it be the serialized CWL document?
Or perhaps it should be the process ID (from the Process table - or something like that)?

chrisburr · 2026-10-22T14:56:35Z

+    if inspect.isasyncgenfunction(plugin):
+        proposals = plugin(inputs=claim(), **kwargs)
+    else:
+        # A synchronous packer cannot await the claim, so it is offered every


I don't think we need to support a synchronous packer, it should just rejected in load_packer

francescobrivio · 2026-10-22T14:58:32Z

My understanding is that this PR is supposed to give the abstractions for DiracX packer functionality. If that is correct, then I think it could come with the most simplistic packer implementation, e.g.: something that generates one parcel for each input, leaving configurations of input pool size, grouping size and output pool size for future implementations.

The "simple packer" is already provided in:
https://github.com/DIRACGrid/diracx/pull/1102/changes#diff-3bd43ea572e0cc3cc5e9d3d70ebeec65594ba7e0abc0f5cdab5a7cc26185397cR26
The fixed_size_packer takes as argument group_size: PositiveInt = 1 and creates parcels with group_size inputs.

chrisburr · 2026-10-22T14:59:55Z

I think it could come with the most simplistic packer implementation

Having this only contain TrivialPacker makes sense to me. It's a useful real packer for things like simulation where you just take a seed.

chrisburr · 2026-10-22T15:15:26Z

I didn't finish reviewing but:

One transaction for the whole run. The ADR says the core "materialises each batch of them … in one transaction" and, under Rationale, "lets the core commit in batches, so a large packing run is naturally chunked and restartable". The PR batches its statements (PARCEL_BATCH_SIZE) but runs everything in the task's single injected transaction. That means:
- the row locks on up to 10k claimed inputs are held for the whole run, which makes the gap-lock contention with the feeder worse;
- once Worker commits the injected DB transaction when a task fails #1101 is fixed, a failure near the end throws away the whole run.
Requirements merge is shallow. {**template, **proposal.requirements} replaces any nested key wholesale. The ADR only says "merged over the transformation's template". If requirements have nested structure, such as resource ranges, a packer that overrides one sub-key loses the rest. Worth deciding whether shallow is the intended meaning.
feeder_active when there's no feeder. feeder_active = feeder is not None and feeder_enabled. A transformation whose Feeder is None ("inputs arrive by other means", per the schema docstring) therefore always looks drained. FixedSize then packs a short remainder on every sweep. That may be intended, but it's a different meaning from "the feeder is off".

10k claimed inputs seems inappropriately small from my past operation experience. We should think of it in terms of memory overhead as that's the main factor.

mmascher and others added 7 commits October 19, 2026 16:51

Merge branch 'DIRACGrid:main' into feat/1083-transformation-system-pa…

0a6540d

…cker

Copilot AI balanced review requested due to automatic review settings October 21, 2026 08:37

mmascher marked this pull request as draft October 21, 2026 08:38

Copilot started reviewing on behalf of mmascher October 21, 2026 08:38

View session

Copilot AI reviewed Oct 21, 2026

View reviewed changes

Rename RequirementsID to RequirementsTemplateID

2950ea3

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

francescobrivio reviewed Oct 21, 2026

View reviewed changes

chrisburr reviewed Oct 22, 2026

View reviewed changes

amaltaro reviewed Oct 22, 2026

View reviewed changes

chrisburr reviewed Oct 22, 2026

View reviewed changes

	return [MutexLock(TRANSFORMATION, self.transformation_id)]
	return [MutexLock(TRANSFORMATION, self.transformation_id, "packer")]



		def test_packer_input_is_frozen():
		packer_input = PackerInput(input_id=uuid4(), lfn="/vo/file")

Conversation

mmascher commented Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

What's included

Scope decisions

Testing

Open points

Uh oh!

mmascher commented Oct 21, 2026

Uh oh!

read-the-docs-community Bot commented Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Documentation build overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

francescobrivio left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

amaltaro commented Oct 22, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

francescobrivio commented Oct 22, 2026

Uh oh!

chrisburr commented Oct 22, 2026

Uh oh!

chrisburr commented Oct 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

mmascher commented Oct 21, 2026 •

edited

Loading

read-the-docs-community Bot commented Oct 21, 2026 •

edited

Loading