feat(grzctl,grz-common)!: add `grzctl process` + streaming backend by tedil · Pull Request #513 · BfArM-MVH/grz-tools

tedil · 2026-02-09T16:45:43Z

Resolves #512

TODO

Tasks

Replace this entire PR description with the desired final squash commit message.
- Make sure to follow the release-please format by placing these at the end, separated by newlines.

Multiple Conventional Commit statements can be included in the description, in addition to the PR title.

Hoeze · 2026-02-09T16:47:22Z

@copilot please review

Copilot · 2026-02-09T16:47:31Z

@Hoeze I've opened a new pull request, #514, to work on those changes. Once the pull request is ready, I'll request review from you.

Copilot

Pull request overview

Adds a new grzctl process command backed by a modular streaming pipeline (S3 download → Crypt4GH decrypt → optional validation → Crypt4GH encrypt → S3 multipart upload), plus redaction and integration tests to compare results with the manual CLI workflow.

Changes:

Introduces grzctl process command, config model, example config, and CLI wiring.
Adds grz_common.pipeline streaming components (S3 source/sink, Crypt4GH stages, validators, operations utilities) and a StreamingPipelineWorker.
Adds integration/unit tests for streaming pipeline behavior and CLI process flow.

Reviewed changes

Copilot reviewed 32 out of 34 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
tests/test_streaming_pipeline_comparison.py	Adds integration tests comparing manual CLI steps vs streaming decrypt behavior.
tests/cli/test_process.py	Adds integration tests for `grzctl process` with mocked S3 and DB, including failure cases.
pyproject.toml	Adds additional dependencies and updates tooling excludes/ignores.
packages/grzctl/src/grzctl/models/config.py	Adds `ProcessConfig` and S3 multi-inbox support models.
packages/grzctl/src/grzctl/commands/process.py	Implements the `grzctl process` command orchestration.
packages/grzctl/src/grzctl/cli.py	Registers the new `process` command.
packages/grzctl/examples/config.process.yaml	Provides an example configuration for streaming processing.
packages/grz-pydantic-models/src/grz_pydantic_models/submission/metadata/v1.py	Adds metadata redaction helpers for logs + archived metadata.
packages/grz-db/src/grz_db/models/submission.py	Extends submission state enum for the new processing flow.
packages/grz-db/src/grz_db/models/base.py	Removes a magic-method docstring ignore and enforces subclass requirements.
packages/grz-common/tests/pipeline/test_pipeline_components.py	Adds unit tests for decompression and validation pipeline components.
packages/grz-common/tests/pipeline/test_crypt4gh_components.py	Adds unit tests for streaming Crypt4GH encryption/decryption components.
packages/grz-common/src/grz_common/workers/upload.py	Switches uploads to use streaming `UploadOperation` instead of boto3 transfer helpers.
packages/grz-common/src/grz_common/workers/streaming.py	Adds the streaming worker that runs the end-to-end pipeline and uploads logs/metadata.
packages/grz-common/src/grz_common/workers/download.py	Switches downloads to use streaming `DownloadOperation` instead of boto3 transfer helpers.
packages/grz-common/src/grz_common/utils/redaction.py	Adds file redaction helper used before archiving logs.
packages/grz-common/src/grz_common/utils/crypt.py	Refactors file encrypt/decrypt helpers to use the streaming operations wrappers.
packages/grz-common/src/grz_common/progress/states.py	Adds `ProcessingState` for streaming pipeline progress logging.
packages/grz-common/src/grz_common/progress/init.py	Exposes `ProcessingState` from the progress package.
packages/grz-common/src/grz_common/pipeline/validators.py	Introduces streaming validators (checksum, FASTQ, BAM).
packages/grz-common/src/grz_common/pipeline/utils.py	Adds signal handling + helper utilities for staged pipeline cleanup.
packages/grz-common/src/grz_common/pipeline/s3_keys.py	Centralizes S3 key naming for inbox/archive structures.
packages/grz-common/src/grz_common/pipeline/s3.py	Adds streaming S3 downloader and multipart uploader stages.
packages/grz-common/src/grz_common/pipeline/processing.py	Implements the tee’d streaming pipeline orchestrator used by `process`.
packages/grz-common/src/grz_common/pipeline/operations.py	Adds file-based wrappers around streaming stages for CLI subcommands.
packages/grz-common/src/grz_common/pipeline/crypt4gh.py	Adds streaming Crypt4GH encrypt/decrypt transformers.
packages/grz-common/src/grz_common/pipeline/constants.py	Re-exports multipart constants for pipeline internals.
packages/grz-common/src/grz_common/pipeline/compressors.py	Adds gzip decompressor with optional ISA-L acceleration.
packages/grz-common/src/grz_common/pipeline/base.py	Adds pipeline stage interfaces and shared context.
packages/grz-common/src/grz_common/pipeline/init.py	Exports the modular pipeline public API.
packages/grz-common/src/grz_common/constants.py	Adds multipart/upload constants and tqdm-safe logging handler.
packages/grz-common/pyproject.toml	Adds optional `fast` extra for ISA-L gzip acceleration.

Comments suppressed due to low confidence (2)

tests/cli/test_process.py:1

The test claims to trigger FASTQ validation failure, but it uploads non-Crypt4GH content with a .c4gh suffix. The streaming pipeline will fail earlier during Crypt4GH header parsing/decryption (so this test is likely asserting the wrong failure mode). Encrypt gzipped_content with Crypt4GH using the inbox/public key before uploading, or change the test expectations to assert decryption/header failure instead of validation failure.
pyproject.toml:1
boto3-stubs is typically a typing/dev-only dependency (large optional dependency surface), and isal is a platform-dependent C extension that can break installs on unsupported environments. Consider moving boto3-stubs to a dev dependency group and keeping isal behind an optional extra (you already introduced grz-common[fast]), rather than making it a required runtime dependency at the workspace root.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Hoeze · 2026-02-12T14:17:47Z

+        self.threads = threads
+        self.max_concurrent_uploads = max_concurrent_uploads
+
+        self.pool_size = max(10, threads * (1 + max_concurrent_uploads) + 1)


what is the logic behind that?

default pool size is 10, urllib will drop connections; not sure this is really needed (at least not if we stay below 10 ;) )

A minor thing is to make increased use of readable constants. More importantly an upper bound is not clear - is that a concern? Otherwise it could be something like:
MIN_POOL_SIZE=10
MAX_POOL_SIZE=15
...
self.pool_size = min(MAX_POOL_SIZE, max(MIN_POOL_SIZE, ... ))

…l process

…public

tedil · 2026-06-02T14:13:26Z

TODO: replace from nacl.public import PrivateKey similar to #572

…stance

beoinformatics

impressive work! I am only finding 2 issues, none of them critical:

in case of an error during file copying, the cleanup may not be complete (no abort on server side, also no deletion of partial archive data but that may be by design)
ETag is in my view doppelt-gemoppelt with MD5 comparison and since ETags depend on server side settings I my vote would be not using them at all

beoinformatics · 2026-06-10T13:10:58Z

+                self.close()
+            # Sinks like open files or DevNullSink need closing to ensure flush
+            other.close()
+        return other


One potential issue: If the copy errors (or the validators raise on self.close()), the upload is neither committed nor aborted — it may leak an incomplete multipart upload that sits on the server side (Ceph). Worth aborting on the failure path. This issue is verified in a unit test that will be part of another branch.

Yes, good point, in general we should go through the code and ensure we properly catch errors/exceptions and handle such cases gracefully, including multipart abort etc.

beoinformatics · 2026-06-10T13:28:33Z

+        )
+
+        server_etag = resp["ETag"].strip('"')
+        if server_etag != local_md5_hex:


do we need ETag logic at all? We already compare MD5 values, so if feels like 'belt and suspenders'. Plus server-side ETag values cannot be fully relied upon, they may or may not be set to the MD5 value. My vote would be to remove the ETag logic

No, since this is also used during archival, ETags are vitally important.

beoinformatics · 2026-06-10T13:54:09Z

+        5. Optionally clean the inbox.
+
+        :param submission_metadata: The parsed metadata object containing donor and file information.
+        :raises RuntimeError: If consistency checks fail or any file fails validation.


just flagging to ensure that the decision is consciously made that in case of errors we keep files uploaded so far on the archive.

This is one of the points that are open for discussion, let's put that on the agenda for our next meeting

…l target archive

Hoeze · 2026-06-17T14:18:53Z

+            log.info("Interrogation config keep_failed is True. Leaving failed files in interrogation bucket.")
+            return
+
+        log.info("Cleaning up interrogation bucket due to failure...")


log.warning()?

Problem: When the process pipeline fails partway through writing a file to the archive, that file's S3 multipart upload was previously left open - its already-staged parts lingered as orphaned upload fragments that never became a usable object. Fix: Now, on failure, that file's upload is aborted and its staged parts are discarded. Test: A new unit test reproduces the original bug and verifies the fix. --------- Co-authored-by: Eckart Bindewald <eckart.bindewald@uni-tuebingen.de> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tedil requested a review from Hoeze February 9, 2026 16:45

Copilot AI mentioned this pull request Feb 9, 2026

[WIP] Add grzctl process and streaming backend #514

Closed

Hoeze requested a review from Copilot February 9, 2026 16:48

Copilot AI reviewed Feb 9, 2026

View reviewed changes

Copilot AI mentioned this pull request Feb 10, 2026

[WIP] WIP Address feedback from PR #513 on grzctl process and streaming backend #515

Closed

BfArM-MVH deleted a comment from Copilot AI Feb 10, 2026

Hoeze reviewed Feb 12, 2026

View reviewed changes

Hoeze force-pushed the feat/grzctl-process branch 3 times, most recently from 9940111 to d98cec5 Compare February 13, 2026 14:22

tedil force-pushed the feat/grzctl-process branch 2 times, most recently from c085476 to 34fe475 Compare February 23, 2026 14:19

tedil and others added 16 commits March 14, 2026 23:33

error on 0 read fastq files as well

32fd2a1

cast to bytes

4701ffc

add internal buffer to Crypt4GHEncryptor just to be on the safe side

eac59fe

fix lints

bf8a446

revert accidental change

ed60cd8

remove unused param

139fb07

refactor config structure for grzctl process

821d36f

update example config.yaml for grzctl process

fcef6a4

update tests/mocks to reflect grzctl process config structure changes

838c2bc

allow explicit specification of passphrase in retrieve_private_key

4f289ad

.

48d349d

add private_key_passphrase explicitly to inboxtarget models for grzct…

b6ee2c2

…l process

update tests

31007f4

add tests for passphrase override via json or .env file

289623d

.

cccb8ce

Explicit ReadStream and WriteStream

4975394

tedil and others added 18 commits May 12, 2026 10:17

merge origin/main

033f77c

Merge remote-tracking branch 'origin/main' into feat/grzctl-process

0a326f0

fix decryption

787f1a2

fix mean read length check in grz-check

94037aa

fix fastq check tests

0451f08

fix c4gh key generate missing comment

de02792

fmt

9049eef

exit if no session keys defined

e6a8260

adapt crypt4gh interactive passphrase test to catch systemexit instead

d00a9ac

fix type hint

270fcc5

update uv lock

1c99eb7

merge origin/main

e830802

fix tuple in IgnoringBaseSettings.from_path

0a8fc92

use cryptography.hazmat.primitives.* instead of PrivateKey from nacl.…

e6f857a

…public

use crypt4gh's custom libsodium bindings directly

63611d2

don't read files completely into memory…

830d2ef

merge origin/main

93cf477

Merge branch 'main' into feat/grzctl-process

3454105

tedil added 2 commits June 8, 2026 14:18

Merge remote-tracking branch 'origin/main' into feat/grzctl-process

0eb95d7

get pruefbericht from db instead of from encrypted submission mock in…

39ceffb

…stance

beoinformatics requested changes Jun 10, 2026

View reviewed changes

tedil added 5 commits June 17, 2026 15:16

Merge remote-tracking branch 'origin/main' into feat/grzctl-process

c74fff9

introduce interrogation bucket and s3 copy from interrogation to fina…

8c1742f

…l target archive

don't track keys, use known-in-advance keys directly

56d2b73

address lints/typecheck

cdc6ad1

update high-level description docs

6c47880

Hoeze reviewed Jun 17, 2026

View reviewed changes

beoinformatics and others added 2 commits June 23, 2026 13:00

merge origin/main

335201b

Uh oh!

Conversation

tedil commented Feb 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

TODO

Tasks

Uh oh!

Hoeze commented Feb 9, 2026

Uh oh!

Copilot AI commented Feb 9, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tedil commented Jun 2, 2026

Uh oh!

beoinformatics 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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

tedil commented Feb 9, 2026 •

edited

Loading