diff --git a/.github/workflows/cd-monorepo.yml b/.github/workflows/cd-monorepo.yml index 977c7556024..b27ceb12881 100644 --- a/.github/workflows/cd-monorepo.yml +++ b/.github/workflows/cd-monorepo.yml @@ -146,22 +146,38 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + # The branch already holds the version to release. The bump happens after + # the publish, so the version on the branch names an unreleased version + - name: Read the version to release + run: | + git pull + VERSION=$(python3 syft_client/version.py) + echo "VERSION=$VERSION" >> $GITHUB_ENV + echo "Releasing syft-client $VERSION" + + # Artifacts live inside the package, so they must be committed before the + # build. Re-running the export writes nothing when they already exist. + - name: Freeze release artifacts + run: | + just export-release-artifacts + git add syft_client/migrations/history + git diff --cached --quiet || \ + git commit -m "Freeze syft-client v${{ env.VERSION }} release artifacts" + - name: Upload to PyPI id: publish env: TWINE_USERNAME: __token__ TWINE_PASSWORD: ${{ secrets.PYPI_PASS_SYFT_CLIENT }} run: | - git pull - just bump-and-publish ${{ inputs.bump_type }} - VERSION=$(python3 syft_client/version.py) - echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "version=$VERSION" >> $GITHUB_OUTPUT + just publish + echo "version=${{ env.VERSION }}" >> $GITHUB_OUTPUT - # bump and publish already does committing - - name: Push changes to syft-client repo + # The tag must name the published version, so it is created before the bump. + - name: Tag the release, then bump for the next one run: | git tag "syft-client/v${{ env.VERSION }}" + just bump ${{ inputs.bump_type }} git push origin --follow-tags post-release-tests: diff --git a/.github/workflows/cd-syft-bg.yml b/.github/workflows/cd-syft-bg.yml index 365c55e85f5..e30fdf4aa7d 100644 --- a/.github/workflows/cd-syft-bg.yml +++ b/.github/workflows/cd-syft-bg.yml @@ -43,16 +43,15 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - - name: Bump version + # The branch already holds the version to release. The bump happens after + # the publish, so the version on the branch always names an unreleased + # version while people develop. + - name: Read the version to release run: | git pull - pip install packaging - OUTPUT=$(python scripts/bump_version.py syft-bg ${{ inputs.bump_type }}) - VERSION=$(echo "$OUTPUT" | sed -n '1p') - MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + VERSION=$(python -c "import tomllib; print(tomllib.load(open('packages/syft-bg/pyproject.toml','rb'))['project']['version'])") echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "MODIFIED=$MODIFIED" >> $GITHUB_ENV - echo "Bumped syft-bg to $VERSION (modified: $MODIFIED)" + echo "Releasing syft-bg $VERSION" - name: Build package working-directory: packages/syft-bg @@ -65,9 +64,14 @@ jobs: TWINE_PASSWORD: ${{ secrets.PYPI_PASS_SYFT_BG }} run: uvx twine upload --verbose dist/* - - name: Commit and tag + # The tag must name the published version, so it is created before the bump. + - name: Tag the release, then bump for the next one run: | - git add ${{ env.MODIFIED }} - git commit -m "Release syft-bg v${{ env.VERSION }}" git tag "syft-bg/v${{ env.VERSION }}" + pip install packaging + OUTPUT=$(python scripts/bump_version.py syft-bg ${{ inputs.bump_type }} --dependents published) + NEXT=$(echo "$OUTPUT" | sed -n '1p') + MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + git add $MODIFIED + git commit -m "Bump syft-bg to $NEXT for the next release" git push origin --follow-tags diff --git a/.github/workflows/cd-syft-dataset.yml b/.github/workflows/cd-syft-dataset.yml index 8b572684918..f1a88f369c0 100644 --- a/.github/workflows/cd-syft-dataset.yml +++ b/.github/workflows/cd-syft-dataset.yml @@ -43,16 +43,24 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - - name: Bump version + # The branch already holds the version to release. The bump happens after + # the publish, so the version on the branch always names an unreleased + # version while people develop. + - name: Read the version to release run: | git pull - pip install packaging - OUTPUT=$(python scripts/bump_version.py syft-dataset ${{ inputs.bump_type }}) - VERSION=$(echo "$OUTPUT" | sed -n '1p') - MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + VERSION=$(python -c "import tomllib; print(tomllib.load(open('packages/syft-datasets/pyproject.toml','rb'))['project']['version'])") echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "MODIFIED=$MODIFIED" >> $GITHUB_ENV - echo "Bumped syft-dataset to $VERSION (modified: $MODIFIED)" + echo "Releasing syft-dataset $VERSION" + + # Artifacts live inside the package, so they must be committed before the + # build. Re-running the export writes nothing when they already exist. + - name: Freeze release artifacts + run: | + uv run python packages/syft-datasets/scripts/export_release_artifact.py + git add packages/syft-datasets/src/syft_datasets/migrations/history + git diff --cached --quiet || \ + git commit -m "Freeze syft-dataset v${{ env.VERSION }} release artifacts" - name: Build package working-directory: packages/syft-datasets @@ -65,9 +73,14 @@ jobs: TWINE_PASSWORD: ${{ secrets.PYPI_PASS_SYFT_DATASET }} run: uvx twine upload --verbose dist/* - - name: Commit and tag + # The tag must name the published version, so it is created before the bump. + - name: Tag the release, then bump for the next one run: | - git add ${{ env.MODIFIED }} - git commit -m "Release syft-dataset v${{ env.VERSION }}" git tag "syft-dataset/v${{ env.VERSION }}" + pip install packaging + OUTPUT=$(python scripts/bump_version.py syft-dataset ${{ inputs.bump_type }} --dependents published) + NEXT=$(echo "$OUTPUT" | sed -n '1p') + MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + git add $MODIFIED + git commit -m "Bump syft-dataset to $NEXT for the next release" git push origin --follow-tags diff --git a/.github/workflows/cd-syft-job.yml b/.github/workflows/cd-syft-job.yml index a9bdb706b2a..bba4fc0c342 100644 --- a/.github/workflows/cd-syft-job.yml +++ b/.github/workflows/cd-syft-job.yml @@ -43,16 +43,24 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - - name: Bump version + # The branch already holds the version to release. The bump happens after + # the publish, so the version on the branch always names an unreleased + # version while people develop. + - name: Read the version to release run: | git pull - pip install packaging - OUTPUT=$(python scripts/bump_version.py syft-job ${{ inputs.bump_type }}) - VERSION=$(echo "$OUTPUT" | sed -n '1p') - MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + VERSION=$(python -c "import tomllib; print(tomllib.load(open('packages/syft-job/pyproject.toml','rb'))['project']['version'])") echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "MODIFIED=$MODIFIED" >> $GITHUB_ENV - echo "Bumped syft-job to $VERSION (modified: $MODIFIED)" + echo "Releasing syft-job $VERSION" + + # Artifacts live inside the package, so they must be committed before the + # build. Re-running the export writes nothing when they already exist. + - name: Freeze release artifacts + run: | + uv run python packages/syft-job/scripts/export_release_artifact.py + git add packages/syft-job/src/syft_job/migrations/history + git diff --cached --quiet || \ + git commit -m "Freeze syft-job v${{ env.VERSION }} release artifacts" - name: Build package working-directory: packages/syft-job @@ -65,9 +73,14 @@ jobs: TWINE_PASSWORD: ${{ secrets.PYPI_PASS_SYFT_JOB }} run: uvx twine upload --verbose dist/* - - name: Commit and tag + # The tag must name the published version, so it is created before the bump. + - name: Tag the release, then bump for the next one run: | - git add ${{ env.MODIFIED }} - git commit -m "Release syft-job v${{ env.VERSION }}" git tag "syft-job/v${{ env.VERSION }}" + pip install packaging + OUTPUT=$(python scripts/bump_version.py syft-job ${{ inputs.bump_type }} --dependents published) + NEXT=$(echo "$OUTPUT" | sed -n '1p') + MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + git add $MODIFIED + git commit -m "Bump syft-job to $NEXT for the next release" git push origin --follow-tags diff --git a/.github/workflows/cd-syft-permissions.yml b/.github/workflows/cd-syft-permissions.yml index f34b0c686d9..49e01b19977 100644 --- a/.github/workflows/cd-syft-permissions.yml +++ b/.github/workflows/cd-syft-permissions.yml @@ -43,16 +43,15 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - - name: Bump version + # The branch already holds the version to release. The bump happens after + # the publish, so the version on the branch always names an unreleased + # version while people develop. + - name: Read the version to release run: | git pull - pip install packaging - OUTPUT=$(python scripts/bump_version.py syft-permissions ${{ inputs.bump_type }}) - VERSION=$(echo "$OUTPUT" | sed -n '1p') - MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + VERSION=$(python -c "import tomllib; print(tomllib.load(open('packages/syft-permissions/pyproject.toml','rb'))['project']['version'])") echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "MODIFIED=$MODIFIED" >> $GITHUB_ENV - echo "Bumped syft-permissions to $VERSION (modified: $MODIFIED)" + echo "Releasing syft-permissions $VERSION" - name: Build package working-directory: packages/syft-permissions @@ -65,9 +64,14 @@ jobs: TWINE_PASSWORD: ${{ secrets.PYPI_PASS_SYFT_PERMISSIONS }} run: uvx twine upload --verbose dist/* - - name: Commit and tag + # The tag must name the published version, so it is created before the bump. + - name: Tag the release, then bump for the next one run: | - git add ${{ env.MODIFIED }} - git commit -m "Release syft-permissions v${{ env.VERSION }}" git tag "syft-permissions/v${{ env.VERSION }}" + pip install packaging + OUTPUT=$(python scripts/bump_version.py syft-permissions ${{ inputs.bump_type }} --dependents published) + NEXT=$(echo "$OUTPUT" | sed -n '1p') + MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + git add $MODIFIED + git commit -m "Bump syft-permissions to $NEXT for the next release" git push origin --follow-tags diff --git a/.github/workflows/cd-syft-perms.yml b/.github/workflows/cd-syft-perms.yml index f094d99e176..c8ba7d34902 100644 --- a/.github/workflows/cd-syft-perms.yml +++ b/.github/workflows/cd-syft-perms.yml @@ -43,16 +43,15 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - - name: Bump version + # The branch already holds the version to release. The bump happens after + # the publish, so the version on the branch always names an unreleased + # version while people develop. + - name: Read the version to release run: | git pull - pip install packaging - OUTPUT=$(python scripts/bump_version.py syft-perms ${{ inputs.bump_type }}) - VERSION=$(echo "$OUTPUT" | sed -n '1p') - MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + VERSION=$(python -c "import tomllib; print(tomllib.load(open('packages/syft-perms/pyproject.toml','rb'))['project']['version'])") echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "MODIFIED=$MODIFIED" >> $GITHUB_ENV - echo "Bumped syft-perms to $VERSION (modified: $MODIFIED)" + echo "Releasing syft-perms $VERSION" - name: Build package working-directory: packages/syft-perms @@ -65,9 +64,14 @@ jobs: TWINE_PASSWORD: ${{ secrets.PYPI_PASS_SYFT_PERMS }} run: uvx twine upload --verbose dist/* - - name: Commit and tag + # The tag must name the published version, so it is created before the bump. + - name: Tag the release, then bump for the next one run: | - git add ${{ env.MODIFIED }} - git commit -m "Release syft-perms v${{ env.VERSION }}" git tag "syft-perms/v${{ env.VERSION }}" + pip install packaging + OUTPUT=$(python scripts/bump_version.py syft-perms ${{ inputs.bump_type }} --dependents published) + NEXT=$(echo "$OUTPUT" | sed -n '1p') + MODIFIED=$(echo "$OUTPUT" | sed -n '2p') + git add $MODIFIED + git commit -m "Bump syft-perms to $NEXT for the next release" git push origin --follow-tags diff --git a/.github/workflows/post-release-tests.yml b/.github/workflows/post-release-tests.yml index 2b71535e4d7..16229a3bd73 100644 --- a/.github/workflows/post-release-tests.yml +++ b/.github/workflows/post-release-tests.yml @@ -62,3 +62,8 @@ jobs: run: | source .venv/bin/activate pytest -n auto ./tests/unit + + - name: Run client migration tests + run: | + source .venv/bin/activate + pytest -n auto ./tests/migrations diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml index bce442931be..41ae0f7fdb6 100644 --- a/.github/workflows/unit-tests.yml +++ b/.github/workflows/unit-tests.yml @@ -198,3 +198,6 @@ jobs: - name: Run migration tests run: just test-unit-migration + + - name: Run client migration tests + run: just test-client-migrations diff --git a/Justfile b/Justfile index 74fcb344912..b8bae8d4b8c 100644 --- a/Justfile +++ b/Justfile @@ -10,7 +10,6 @@ _nc := '\033[0m' alias b := build alias p := publish -alias bp:= bump-and-publish # --------------------------------------------------------------------------------------------------------------------- @@ -38,6 +37,10 @@ test-unit-migration: #!/bin/bash uv run pytest -n auto ./packages/syft-migration/tests +test-client-migrations: + #!/bin/bash + uv run pytest -n auto ./tests/migrations + test-unit-enclave: #!/bin/bash @@ -136,12 +139,10 @@ publish: build uvx twine upload dist/* @echo "{{ _green }}Publish complete!{{ _nc }}" -# Bump version and publish to PyPI +# Export the frozen release artifacts for the current version [group('publish')] -bump-and-publish part="patch": - just bump {{ part }} - just publish - @echo "{{ _green }}Bump and publish complete!{{ _nc }}" +export-release-artifacts: + uv run python scripts/export_release_artifact.py # Launch Jupyter Lab jupyter: diff --git a/docs/release.md b/docs/release.md index 2c842757c5f..be10831da81 100644 --- a/docs/release.md +++ b/docs/release.md @@ -2,16 +2,46 @@ ## Overview -Releases are managed through dedicated release branches. The mono repo release job handles bumping versions and pushing tags for all individual packages automatically. +Releases are managed through dedicated release branches. The mono repo release job handles publishing, tagging and bumping versions for all individual packages automatically. + +## Version order + +A release publishes the version that is **already on the branch**. The release then tags that version. After the tag, the release job bumps the version for the next release. + +The version on a branch is always a version that is **not yet published**. Therefore one version string always refers to one build. + +Do not change a version by hand before a release. The release job makes the bump. ## Steps -1. **Create a release branch** from `main`, dont include the patch version in the semver, so we can hotfix patches on the same branch (e.g. `release/v0.1`). If you are patching, re-use the branch +1. **Create a release branch** from `main`, don't include the patch version in the semver, so we can hotfix patches on the same branch (e.g. `release/v0.1`). If you are patching, re-use the branch. 2. **Run the release workflow.** You can trigger frmo github UI from the Actions tab. In most cases, release the mono repo — this releases all individual packages (`syft-client`, `syft-job`, `syft-dataset`, etc.) in one go. You only need to release individual packages if they are changed, but we are not detecting that automatically currently. 3. **Integration tests are optional.** You can skip them during the release if needed. Unit tests should still pass. -4. **Versions are bumped **before releasing to pypi** and pushed automatically** by the release process — no manual version edits required. +4. **The release job publishes, tags, and then bumps the version.** No manual version edit is necessary. 5. Merge the release branch back into `main` to ensure all version bumps and hotfixes are carried forward. +## Release artifacts + +`syft-client`, `syft-job`, and `syft-dataset` each write a release artifact. The artifact records the object versions of that release. It also records the exact schema of each object version. + +The drift check compares the current models against these files. If an artifact is absent, the drift check has nothing to compare for that version. + +The artifacts are inside the package, so the release job runs the export before the build: + +``` +uv run python scripts/export_release_artifact.py # syft-client +uv run python packages/syft-job/scripts/export_release_artifact.py # syft-job +uv run python packages/syft-datasets/scripts/export_release_artifact.py # syft-dataset +``` + +A developer can also run an export in a pull request. The version on the branch is the version that the next release publishes. The artifact is therefore available for review before the release. + +An artifact is permanent. If an artifact for a version exists, a second export writes nothing and reports success. + +An export stops with an error if the protocol changed but the protocol version constant did not change. The error message gives the name of the constant to bump. + +The drift check has one known limit. A new protocol generation adds object versions, and no artifact freezes those versions until the release of that generation. The drift check therefore cannot see a change to them. Frequent releases keep this period short. + ## Hotfixes If a fix is needed after cutting the release branch, apply the hotfix directly to the release branch and re-release from there. diff --git a/packages/syft-datasets/scripts/export_release_artifact.py b/packages/syft-datasets/scripts/export_release_artifact.py index f6a97db3bbd..2e35ebbb16c 100644 --- a/packages/syft-datasets/scripts/export_release_artifact.py +++ b/packages/syft-datasets/scripts/export_release_artifact.py @@ -3,6 +3,9 @@ Run on EVERY release (uv run python scripts/export_release_artifact.py): always writes the package release info; additionally writes the protocol artifact when this release introduces a new protocol version. + +Artifacts are frozen once written. Running this again for the same version +writes nothing and succeeds, so a release can re-run it safely. """ import sys @@ -16,6 +19,14 @@ def main() -> None: # Import the models so every versioned object is registered. import syft_datasets # noqa: F401 + if dataset_registry.protocol_bump_missing(): + sys.exit( + "The dataset protocol changed since the released " + f"protocol-{dataset_registry.latest_released_protocol_version()}.json; " + "bump DATASET_PROTOCOL_VERSION in " + "syft_datasets/migrations/registry.py before releasing." + ) + if dataset_registry.protocol_changed_without_bump(): sys.exit( "The dataset protocol changed compared to the released " @@ -28,11 +39,17 @@ def main() -> None: PROTOCOLS_DIR.mkdir(parents=True, exist_ok=True) info_path = PACKAGE_ARTIFACTS_DIR / f"syft-dataset-{__version__}.json" - dataset_registry.compute_released_package_protocol_info().save(info_path) - print(f"Wrote {info_path}") - protocol_path = PROTOCOLS_DIR / f"protocol-{DATASET_PROTOCOL_VERSION}.json" - if not protocol_path.exists(): + + if info_path.exists(): + print(f"Package artifact already present: {info_path}") + else: + dataset_registry.compute_released_package_protocol_info().save(info_path) + print(f"Wrote {info_path}") + + if protocol_path.exists(): + print(f"Protocol artifact already present: {protocol_path}") + else: dataset_registry.compute_released_protocol().save(protocol_path) print(f"Wrote {protocol_path} (new protocol version)") diff --git a/packages/syft-datasets/src/syft_datasets/dataset_storage.py b/packages/syft-datasets/src/syft_datasets/dataset_storage.py index fe06581ad3d..fef85ca65a4 100644 --- a/packages/syft-datasets/src/syft_datasets/dataset_storage.py +++ b/packages/syft-datasets/src/syft_datasets/dataset_storage.py @@ -1,3 +1,4 @@ +import logging import shutil from dataclasses import dataclass, field from datetime import datetime, timezone @@ -27,6 +28,8 @@ from .protocolcodecs import CODECS, ProtocolCodec from .url import SyftBoxURL +logger = logging.getLogger(__name__) + __all__ = [ "DatasetRef", "DatasetNotFoundError", @@ -138,17 +141,29 @@ def negotiated_protocol_version_for_peer( """The dataset protocol version to speak with ``peer_email``. Negotiated as the minimum of our own protocol version and the peer's, so - both sides use a version they can read. A peer without a known schema - raises by default; with ``raise_on_unknown=False`` it is assumed to run - the current protocol. + both sides use a version they can read. The result must also be at or + above the floor of each side, or the negotiation raises. A peer without a + known schema raises by default; with ``raise_on_unknown=False`` it is + assumed to run the current protocol. """ schema = self.peer_schemas.get(peer_email) if schema is not None: - return min(DATASET_PROTOCOL_VERSION, schema.version, key=int) + return self.registry.negotiate_protocol_version( + peer_version=schema.version, + peer_min=schema.min_supported_version, + ) if raise_on_unknown: raise MigrationError( f"No dataset protocol schema known for peer {peer_email!r}" ) + # raise_on_unknown=False skips the refusal of a peer with an unknown + # version. A peer that speaks an earlier protocol does not read this + # layout. The dataset never arrives. + logger.warning( + f"No dataset protocol schema known for peer {peer_email!r}. This " + f"client writes dataset protocol {DATASET_PROTOCOL_VERSION}. A peer " + "that speaks an earlier protocol will not read this dataset." + ) return DATASET_PROTOCOL_VERSION def target_protocol_versions_for_peers( @@ -158,8 +173,14 @@ def target_protocol_versions_for_peers( A dataset is written once per distinct version in the audience. A known peer contributes ``min(ours, theirs)``; an unknown peer (or no audience) - contributes the widest-compatible protocol, since we cannot assume it can - read a newer layout. + contributes the widest-compatible protocol, since we cannot assume it + can read a newer layout. + + The two unknown-peer answers differ on purpose. This method serves an + audience. An unknown peer therefore takes the widest protocol, and every + reader can read a copy. ``negotiated_protocol_version_for_peer`` serves + one peer, so an unknown peer takes the current protocol. The caller of + that method accepts the risk when it passes ``raise_on_unknown=False``. """ if not peer_emails: return {self._widest_protocol_version} diff --git a/packages/syft-datasets/src/syft_datasets/migrations/registry.py b/packages/syft-datasets/src/syft_datasets/migrations/registry.py index 661be107d98..396ad7984af 100644 --- a/packages/syft-datasets/src/syft_datasets/migrations/registry.py +++ b/packages/syft-datasets/src/syft_datasets/migrations/registry.py @@ -12,6 +12,11 @@ # syft_datasets folder (see config.protocol_dir_name). DATASET_PROTOCOL_VERSION = "1" +# Oldest dataset protocol this release still reads. "0" refuses no peer. Raise it +# only when the code drops support for a released protocol, because a peer below +# the floor cannot exchange datasets with this release. +MIN_SUPPORTED_DATASET_PROTOCOL_VERSION = "0" + # Package-local registry for all versioned syft-dataset objects. The current # protocol schema is computed from the objects registered into it. dataset_registry = MigrationRegistry( @@ -19,4 +24,5 @@ package_name=PACKAGE_NAME, package_version=__version__, protocol_version=DATASET_PROTOCOL_VERSION, + min_supported_protocol_version=MIN_SUPPORTED_DATASET_PROTOCOL_VERSION, ) diff --git a/packages/syft-datasets/tests/migrations/unit/test_history_artifacts.py b/packages/syft-datasets/tests/migrations/unit/test_history_artifacts.py index be7b2814160..d8d0d90e870 100644 --- a/packages/syft-datasets/tests/migrations/unit/test_history_artifacts.py +++ b/packages/syft-datasets/tests/migrations/unit/test_history_artifacts.py @@ -1,15 +1,14 @@ """The hardcoded release artifacts of past syft-dataset releases.""" +from syft_datasets.migrations import dataset_registry +from syft_datasets.migrations.history import PACKAGE_ARTIFACTS_DIR, PROTOCOLS_DIR +from syft_datasets.models import DatasetV1 from syft_migration import ( MigrationService, ReleasedPackageProtocolInfo, ReleasedProtocol, ) -from syft_datasets.migrations import dataset_registry -from syft_datasets.migrations.history import PACKAGE_ARTIFACTS_DIR, PROTOCOLS_DIR -from syft_datasets.models import DatasetV1 - def test_all_released_package_artifacts_load(): artifact_paths = sorted(PACKAGE_ARTIFACTS_DIR.glob("*.json")) @@ -79,6 +78,16 @@ def test_protocol_bumped_when_changed(): assert not dataset_registry.protocol_changed_without_bump() +def test_protocol_bump_not_missing(): + # Stays live between a protocol bump and the release that freezes it, which is + # exactly where test_protocol_bumped_when_changed goes quiet. + assert not dataset_registry.protocol_bump_missing(), ( + "The dataset protocol changed since the newest released protocol without a " + "bump. Bump DATASET_PROTOCOL_VERSION in " + "syft_datasets/migrations/registry.py, or revert the model change." + ) + + def test_historic_schemas_registered_on_import(): # syft_datasets/__init__ registers every artifact in migrations/history/. assert dataset_registry.package_version_history["0"].version == "0.1.20" diff --git a/packages/syft-job/scripts/export_release_artifact.py b/packages/syft-job/scripts/export_release_artifact.py index e4d054c8ce0..73df7e84997 100644 --- a/packages/syft-job/scripts/export_release_artifact.py +++ b/packages/syft-job/scripts/export_release_artifact.py @@ -3,6 +3,9 @@ Run on EVERY release (uv run python scripts/export_release_artifact.py): always writes the package release info; additionally writes the protocol artifact when this release introduces a new protocol version. + +Artifacts are frozen once written. Running this again for the same version +writes nothing and succeeds, so a release can re-run it safely. """ import sys @@ -16,6 +19,14 @@ def main() -> None: # Import the models so every versioned object is registered. import syft_job # noqa: F401 + if job_registry.protocol_bump_missing(): + sys.exit( + "The job protocol changed since the released " + f"protocol-{job_registry.latest_released_protocol_version()}.json; " + "bump JOB_PROTOCOL_VERSION in syft_job/migrations/registry.py " + "before releasing." + ) + if job_registry.protocol_changed_without_bump(): sys.exit( "The job protocol changed compared to the released " @@ -23,12 +34,21 @@ def main() -> None: "in syft_job/migrations/registry.py before releasing." ) - info_path = PACKAGE_ARTIFACTS_DIR / f"syft-job-{__version__}.json" - job_registry.compute_released_package_protocol_info().save(info_path) - print(f"Wrote {info_path}") + PACKAGE_ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True) + PROTOCOLS_DIR.mkdir(parents=True, exist_ok=True) + info_path = PACKAGE_ARTIFACTS_DIR / f"syft-job-{__version__}.json" protocol_path = PROTOCOLS_DIR / f"protocol-{JOB_PROTOCOL_VERSION}.json" - if not protocol_path.exists(): + + if info_path.exists(): + print(f"Package artifact already present: {info_path}") + else: + job_registry.compute_released_package_protocol_info().save(info_path) + print(f"Wrote {info_path}") + + if protocol_path.exists(): + print(f"Protocol artifact already present: {protocol_path}") + else: job_registry.compute_released_protocol().save(protocol_path) print(f"Wrote {protocol_path} (new protocol version)") diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index 5507841e628..c5a4acdcdc5 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -1,3 +1,4 @@ +import logging from pathlib import Path from typing import Iterator, Optional @@ -15,6 +16,8 @@ from .models import JobState, JobSubmissionMetadata from .protocolcodecs import CODECS, ProtocolCodec +logger = logging.getLogger(__name__) + __all__ = ["JobRef", "JobStateNotFoundError", "JobStorage"] @@ -72,17 +75,29 @@ def negotiated_protocol_version_for_peer( """The job protocol version to speak with ``peer_email``. Negotiated as the minimum of our own protocol version and the peer's, - so both sides use a version they can read. A peer without a known - schema raises by default; with ``raise_on_unknown=False`` it is assumed - to run the current protocol. + so both sides use a version they can read. The result must also be at or + above the floor of each side, or the negotiation raises. A peer without a + known schema raises by default; with ``raise_on_unknown=False`` it is + assumed to run the current protocol. """ schema = self.peer_schemas.get(peer_email) if schema is not None: - return min(JOB_PROTOCOL_VERSION, schema.version, key=int) + return self.registry.negotiate_protocol_version( + peer_version=schema.version, + peer_min=schema.min_supported_version, + ) if raise_on_unknown: raise MigrationError( f"No job protocol schema known for peer {peer_email!r}" ) + # raise_on_unknown=False skips the refusal of a peer with an unknown + # version. A peer that speaks an earlier protocol does not scan this + # layout. It never sees the job. + logger.warning( + f"No job protocol schema known for peer {peer_email!r}. This client " + f"writes job protocol {JOB_PROTOCOL_VERSION}. A peer that speaks an " + "earlier protocol will not see this job." + ) return JOB_PROTOCOL_VERSION def _get_write_target_schema( diff --git a/packages/syft-job/src/syft_job/migrations/registry.py b/packages/syft-job/src/syft_job/migrations/registry.py index 30f527b2d69..039fd03b6f7 100644 --- a/packages/syft-job/src/syft_job/migrations/registry.py +++ b/packages/syft-job/src/syft_job/migrations/registry.py @@ -11,6 +11,11 @@ # jobs under a v segment after the peer email (see config.protocol_dir_name). JOB_PROTOCOL_VERSION = "1" +# Oldest job protocol this release still reads. "0" refuses no peer. Raise it +# only when the code drops support for a released protocol, because a peer below +# the floor cannot exchange jobs with this release. +MIN_SUPPORTED_JOB_PROTOCOL_VERSION = "0" + # Package-local registry for all versioned syft-job objects. The current # protocol schema is computed from the objects registered into it. job_registry = MigrationRegistry( @@ -18,4 +23,5 @@ package_name=PACKAGE_NAME, package_version=__version__, protocol_version=JOB_PROTOCOL_VERSION, + min_supported_protocol_version=MIN_SUPPORTED_JOB_PROTOCOL_VERSION, ) diff --git a/packages/syft-job/tests/migrations/unit/test_history_artifacts.py b/packages/syft-job/tests/migrations/unit/test_history_artifacts.py index 417bb7db4eb..fb759a576cb 100644 --- a/packages/syft-job/tests/migrations/unit/test_history_artifacts.py +++ b/packages/syft-job/tests/migrations/unit/test_history_artifacts.py @@ -87,6 +87,16 @@ def test_protocol_bumped_when_changed(): assert not job_registry.protocol_changed_without_bump() +def test_protocol_bump_not_missing(): + # Stays live between a protocol bump and the release that freezes it, which is + # exactly where test_protocol_bumped_when_changed goes quiet. + assert not job_registry.protocol_bump_missing(), ( + "The job protocol changed since the newest released protocol without a " + "bump. Bump JOB_PROTOCOL_VERSION in syft_job/migrations/registry.py, or " + "revert the model change." + ) + + def test_historic_schemas_registered_on_import(): # syft_job/__init__ registers every artifact in migrations/history/. assert job_registry.package_version_history["0"].version == "0.1.38" diff --git a/packages/syft-migration/src/syft_migration/identity.py b/packages/syft-migration/src/syft_migration/identity.py index b9f2a2d9066..6ab95ab81d4 100644 --- a/packages/syft-migration/src/syft_migration/identity.py +++ b/packages/syft-migration/src/syft_migration/identity.py @@ -23,6 +23,21 @@ def _has_identity(cls: type[MigratableObject]) -> bool: return not (name_field.is_required() or version_field.is_required()) +def _version_order(version: str) -> int: + """Return the sort key of an object version. + + Object versions are incrementing integers held as strings. A string sort puts + ``"10"`` before ``"2"``, so every comparison must use this key. + """ + try: + return int(version) + except ValueError: + raise MigrationError( + f"Object version {version!r} is not an integer. Object versions are " + "incrementing integers, for example '1', '2', '3'." + ) from None + + def _identity(cls: type[MigratableObject]) -> tuple[str, str]: """Return (canonical_name, version) for a concrete subclass. diff --git a/packages/syft-migration/src/syft_migration/registry.py b/packages/syft-migration/src/syft_migration/registry.py index 7c81859ac77..b7dcce22d03 100644 --- a/packages/syft-migration/src/syft_migration/registry.py +++ b/packages/syft-migration/src/syft_migration/registry.py @@ -3,7 +3,12 @@ from collections import deque from typing import TYPE_CHECKING, Callable -from syft_migration.identity import MigrationError, _has_identity, _identity +from syft_migration.identity import ( + MigrationError, + _has_identity, + _identity, + _version_order, +) from syft_migration.schema import ( PackageInfo, ProtocolSchema, @@ -27,11 +32,15 @@ def __init__( package_name: str, package_version: str, protocol_version: str, + min_supported_protocol_version: str = "0", ) -> None: self.protocol_name = protocol_name self.package_name = package_name self.package_version = package_version self.protocol_version = protocol_version + # The oldest protocol version this package still reads. Raise it only + # when the code drops support for a protocol that a release froze. + self.min_supported_protocol_version = min_supported_protocol_version # canonical_name -> {version: object_class} self.objects: dict[str, dict[str, type[MigratableObject]]] = {} # canonical_name -> {(from_version, to_version): migration_fn} @@ -49,6 +58,8 @@ def register_object_version(self, cls: type[MigratableObject]) -> None: if not _has_identity(cls): return canonical_name, version = _identity(cls) + # Reject a version that cannot be ordered, at class definition time. + _version_order(version) existing = self.objects.get(canonical_name, {}).get(version) if existing is not None and existing is not cls: raise MigrationError( @@ -72,7 +83,7 @@ def latest_version(self, canonical_name: str) -> str: versions = self.versions(canonical_name) if not versions: raise MigrationError(f"No versions registered for {canonical_name!r}") - return max(versions) + return max(versions, key=_version_order) # -- migrations -------------------------------------------------------- def register_migration( @@ -205,8 +216,9 @@ def compute_protocol_schema(self) -> ProtocolSchema: return ProtocolSchema( protocol_name=self.protocol_name, version=self.protocol_version, + min_supported_version=self.min_supported_protocol_version, supported_versions={ - canonical_name: sorted(versions) + canonical_name: sorted(versions, key=_version_order) for canonical_name, versions in self.objects.items() }, current_object_schemas={ @@ -217,6 +229,31 @@ def compute_protocol_schema(self) -> ProtocolSchema: }, ) + def negotiate_protocol_version( + self, peer_version: str, peer_min: str | None = None + ) -> str: + """The protocol version to speak with a peer. + + Both sides speak the lower of the two current versions, because each side + must read what the other writes. That version must also be at or above + both floors. A peer that publishes no floor is treated as ``"0"``, which + refuses nothing. + + Raises MigrationError when no version satisfies both sides. + """ + chosen = min(self.protocol_version, peer_version, key=_version_order) + floor = max( + self.min_supported_protocol_version, peer_min or "0", key=_version_order + ) + if _version_order(chosen) < _version_order(floor): + raise MigrationError( + f"No usable {self.protocol_name} protocol version with this peer. " + f"This client speaks {self.protocol_version} and reads down to " + f"{self.min_supported_protocol_version}; the peer speaks " + f"{peer_version} and reads down to {peer_min or '0'}." + ) + return chosen + def compute_released_protocol(self) -> ReleasedProtocol: """The protocol artifact a release emits when the protocol changed.""" return ReleasedProtocol(protocol_schema=self.compute_protocol_schema()) @@ -278,3 +315,25 @@ def protocol_changed_without_bump(self) -> bool: return False current = self.compute_protocol_schema() return released.supported_versions != current.supported_versions + + def latest_released_protocol_version(self) -> str | None: + """The newest protocol version with a frozen schema. None if there is none.""" + if not self.protocol_version_history: + return None + return max(self.protocol_version_history, key=_version_order) + + def protocol_bump_missing(self) -> bool: + """Whether the protocol changed since the newest RELEASED protocol + without a bump of the version constant. + + Only object versions are compared. A protocol change that alters the + on-disk layout, but adds no object version, is invisible here. + """ + latest = self.latest_released_protocol_version() + if latest is None: + return False + released = self.protocol_version_history[latest] + current = self.compute_protocol_schema() + if current.supported_versions == released.supported_versions: + return False + return _version_order(self.protocol_version) <= _version_order(latest) diff --git a/packages/syft-migration/src/syft_migration/schema.py b/packages/syft-migration/src/syft_migration/schema.py index b68bed5ccf4..1230bb7e607 100644 --- a/packages/syft-migration/src/syft_migration/schema.py +++ b/packages/syft-migration/src/syft_migration/schema.py @@ -6,7 +6,7 @@ from pydantic import BaseModel -from syft_migration.identity import MigrationError, _identity +from syft_migration.identity import MigrationError, _identity, _version_order if TYPE_CHECKING: from syft_migration.base import MigratableObject @@ -25,6 +25,9 @@ class ProtocolSchema(BaseModel): # Incrementing protocol version ("0", "1", ...); bumped when the on-disk / # on-the-wire layout of the protocol changes, independent of package versions. version: str + # The oldest protocol version this speaker still reads. A peer that predates + # this field says nothing, so "0" refuses nothing. + min_supported_version: str = "0" # canonical_name -> all supported versions supported_versions: dict[str, list[str]] = {} # canonical_name -> JSON schema of the protocol's current (latest) object @@ -45,13 +48,14 @@ def from_objects( versions = supported_versions.setdefault(canonical_name, []) if object_version not in versions: versions.append(object_version) - if object_version == max(versions): + if object_version == max(versions, key=_version_order): latest_classes[canonical_name] = klass return cls( protocol_name=protocol_name, version=version, supported_versions={ - name: sorted(versions) for name, versions in supported_versions.items() + name: sorted(versions, key=_version_order) + for name, versions in supported_versions.items() }, current_object_schemas={ name: klass.model_json_schema() @@ -64,7 +68,7 @@ def current_schema(self, canonical_name: str) -> str: versions = self.supported_versions.get(canonical_name) if not versions: raise MigrationError(f"Schema does not include object {canonical_name!r}") - return max(versions) + return max(versions, key=_version_order) def save(self, path: PathLike) -> None: Path(path).write_text(self.model_dump_json(indent=2)) diff --git a/packages/syft-migration/tests/test_protocol_floor.py b/packages/syft-migration/tests/test_protocol_floor.py new file mode 100644 index 00000000000..48daf6db75e --- /dev/null +++ b/packages/syft-migration/tests/test_protocol_floor.py @@ -0,0 +1,67 @@ +"""A protocol floor refuses a version that one of the two sides cannot read. + +Both sides publish a floor. Negotiation picks the lower current version, and that +version must be at or above both floors. A floor of "0" refuses nothing. +""" + +import pytest +from syft_migration import MigrationError, MigrationRegistry, ProtocolSchema + + +def _registry(protocol_version: str = "2", floor: str = "0") -> MigrationRegistry: + return MigrationRegistry( + protocol_name="p", + package_name="pkg", + package_version="1.0.0", + protocol_version=protocol_version, + min_supported_protocol_version=floor, + ) + + +def test_schema_floor_defaults_to_zero(): + # A peer that predates the floor field says nothing, so it refuses nothing. + schema = ProtocolSchema(protocol_name="p", version="1") + assert schema.min_supported_version == "0" + + +def test_registry_floor_defaults_to_zero(): + reg = MigrationRegistry( + protocol_name="p", + package_name="pkg", + package_version="1.0.0", + protocol_version="1", + ) + assert reg.min_supported_protocol_version == "0" + + +def test_negotiation_picks_the_lower_version(): + reg = _registry(protocol_version="2") + assert reg.negotiate_protocol_version(peer_version="1") == "1" + assert reg.negotiate_protocol_version(peer_version="3") == "2" + + +def test_negotiation_orders_by_number(): + reg = _registry(protocol_version="10") + assert reg.negotiate_protocol_version(peer_version="9") == "9" + + +def test_our_floor_refuses_an_older_peer(): + reg = _registry(protocol_version="2", floor="2") + with pytest.raises(MigrationError, match="1"): + reg.negotiate_protocol_version(peer_version="1") + + +def test_the_peer_floor_refuses_us(): + reg = _registry(protocol_version="2", floor="0") + with pytest.raises(MigrationError): + reg.negotiate_protocol_version(peer_version="3", peer_min="3") + + +def test_a_zero_floor_on_both_sides_refuses_nothing(): + reg = _registry(protocol_version="5", floor="0") + assert reg.negotiate_protocol_version(peer_version="0", peer_min="0") == "0" + + +def test_an_unknown_peer_floor_is_treated_as_zero(): + reg = _registry(protocol_version="2", floor="0") + assert reg.negotiate_protocol_version(peer_version="1", peer_min=None) == "1" diff --git a/packages/syft-migration/tests/test_release_artifacts.py b/packages/syft-migration/tests/test_release_artifacts.py index 4fb572c3400..19326d73c25 100644 --- a/packages/syft-migration/tests/test_release_artifacts.py +++ b/packages/syft-migration/tests/test_release_artifacts.py @@ -152,3 +152,78 @@ class GadgetV2(MigratableObject, registry=reg): version: str = "2" assert reg.protocol_changed_without_bump() + + +def test_bump_missing_is_live_before_the_protocol_is_released(): + # protocol_changed_without_bump needs a frozen schema for the CURRENT protocol + # version, so it cannot see a change made after a bump. protocol_bump_missing + # compares against the newest released protocol instead. + reg = _fresh_registry(protocol_version="0") + + class WidgetV1(MigratableObject, registry=reg): + canonical_name: str = "widget" + version: str = "1" + + reg.register_released_protocol(released=reg.compute_released_protocol()) + assert reg.latest_released_protocol_version() == "0" + assert not reg.protocol_bump_missing() + + # Bump the protocol, then add an object version. Protocol 1 is not released, + # so the old guard goes quiet and the new one must not. + reg.protocol_version = "1" + + class WidgetV2(MigratableObject, registry=reg): + canonical_name: str = "widget" + version: str = "2" + + assert not reg.protocol_changed_without_bump() + assert not reg.protocol_bump_missing() + + class WidgetV3(MigratableObject, registry=reg): + canonical_name: str = "widget" + version: str = "3" + + # Still one bump ahead of the newest released protocol, so still clean. + assert not reg.protocol_bump_missing() + + # Roll the constant back onto the released protocol: the change is now unbumped. + reg.protocol_version = "0" + assert reg.protocol_bump_missing() + + +def test_bump_missing_compares_against_the_newest_released_protocol(): + reg = _fresh_registry(protocol_version="2") + + class PartV1(MigratableObject, registry=reg): + canonical_name: str = "part" + version: str = "1" + + # Freeze protocol 0 holding only version 1. + reg.register_released_protocol(released=reg.compute_released_protocol()) + protocol_0 = reg.protocol_version_history.pop("2") + protocol_0.version = "0" + reg.register_historic_protocol_schema(schema=protocol_0) + + class PartV2(MigratableObject, registry=reg): + canonical_name: str = "part" + version: str = "2" + + # Freeze protocol 10 holding both versions. A string sort would treat "2" as + # the newest released protocol and miss that the code matches protocol 10. + protocol_10 = reg.compute_released_protocol().protocol_schema + protocol_10.version = "10" + reg.register_historic_protocol_schema(schema=protocol_10) + + assert reg.latest_released_protocol_version() == "10" + assert not reg.protocol_bump_missing() + + +def test_bump_missing_is_false_without_history(): + reg = _fresh_registry() + + class BoltV1(MigratableObject, registry=reg): + canonical_name: str = "bolt" + version: str = "1" + + assert reg.latest_released_protocol_version() is None + assert not reg.protocol_bump_missing() diff --git a/packages/syft-migration/tests/test_version_ordering.py b/packages/syft-migration/tests/test_version_ordering.py new file mode 100644 index 00000000000..c9f3937aafc --- /dev/null +++ b/packages/syft-migration/tests/test_version_ordering.py @@ -0,0 +1,103 @@ +"""Object versions order by number, not as strings.""" + +import pytest + +from syft_migration import ( + MigratableObject, + MigrationError, + MigrationRegistry, + ProtocolSchema, +) + + +def _registry() -> MigrationRegistry: + return MigrationRegistry( + protocol_name="p", + package_name="pkg", + package_version="1.0.0", + protocol_version="1", + ) + + +def _two_digit_registry() -> tuple[ + MigrationRegistry, type[MigratableObject], type[MigratableObject] +]: + """A registry with version 2 and version 10 of the same object.""" + reg = _registry() + + class ThingV2(MigratableObject, registry=reg): + canonical_name: str = "thing" + version: str = "2" + + class ThingV10(MigratableObject, registry=reg): + canonical_name: str = "thing" + version: str = "10" + extra: int = 0 + + return reg, ThingV2, ThingV10 + + +def test_latest_version_orders_by_number(): + reg, _, _ = _two_digit_registry() + assert reg.latest_version(canonical_name="thing") == "10" + + +def test_computed_schema_freezes_the_highest_version(): + # find_schema_drift compares the frozen schema of the highest version. A + # string order freezes version 2 and leaves version 10 unguarded. + reg, _, thing_v10 = _two_digit_registry() + schema = reg.compute_protocol_schema() + assert schema.supported_versions == {"thing": ["2", "10"]} + assert schema.current_object_schemas["thing"] == thing_v10.model_json_schema() + + +def test_current_schema_orders_by_number(): + schema = ProtocolSchema( + protocol_name="p", + version="1", + supported_versions={"thing": ["2", "10"]}, + ) + assert schema.current_schema(canonical_name="thing") == "10" + + +@pytest.mark.parametrize("reverse", [False, True]) +def test_from_objects_picks_the_highest_version(reverse): + _, thing_v2, thing_v10 = _two_digit_registry() + classes = [thing_v10, thing_v2] if reverse else [thing_v2, thing_v10] + schema = ProtocolSchema.from_objects( + protocol_name="p", + version="1", + classes=classes, + ) + assert schema.supported_versions == {"thing": ["2", "10"]} + assert schema.current_object_schemas["thing"] == thing_v10.model_json_schema() + + +def test_upgradeable_path_targets_the_highest_version(): + reg, _, _ = _two_digit_registry() + + # Version 3 has no migration, so it cannot reach version 10. A string order + # makes version 3 the latest and reports the path as trivially available. + class ThingV3(MigratableObject, registry=reg): + canonical_name: str = "thing" + version: str = "3" + + reg.register_migration( + canonical_name="thing", + from_version="2", + to_version="10", + fn=lambda obj: obj, + ) + assert reg.has_upgradeable_path_to_latest(canonical_name="thing", from_version="2") + assert not reg.has_upgradeable_path_to_latest( + canonical_name="thing", from_version="3" + ) + + +def test_non_numeric_object_version_is_rejected(): + reg = _registry() + with pytest.raises(MigrationError): + + class ThingV1Patch(MigratableObject, registry=reg): + canonical_name: str = "thing" + version: str = "1.0" diff --git a/scripts/bump_version.py b/scripts/bump_version.py index 0ad7467e959..4ddd90f3143 100644 --- a/scripts/bump_version.py +++ b/scripts/bump_version.py @@ -1,17 +1,32 @@ -"""Bump a package version and propagate the change to all dependents. +"""Bump the version of one package, and update the packages that depend on it. -Usage: python scripts/bump_version.py +Usage: + python scripts/bump_version.py + [--dependents {bumped,published}] -Output (two lines): - Line 1: new version - Line 2: space-separated list of all modified pyproject.toml files +The script writes the new version into the pyproject.toml of the package. It +then writes a version pin for the package into each pyproject.toml that depends +on it. + +The --dependents option selects the version for those pins: + +- published: the version that was in the file before this run. A release + publishes the version on the branch, and bumps the version after that. This + version is therefore the version on PyPI. Use this option for a release. +- bumped: the new version. PyPI does not have this version yet. Use this option + only if the script runs before the release. + +The script prints two lines: + +- Line 1: the new version. +- Line 2: the modified pyproject.toml files, separated by spaces. """ import argparse import re -import tomllib from pathlib import Path +import tomllib from packaging.version import Version REPO_ROOT = Path(__file__).resolve().parent.parent @@ -88,11 +103,24 @@ def main() -> None: ) parser.add_argument("package_name", help="Package name (e.g. syft-perms)") parser.add_argument("bump_type", choices=["major", "minor", "patch"]) + parser.add_argument( + "--dependents", + choices=["bumped", "published"], + default="bumped", + help=( + "Version for the dependent pins. 'bumped' is the new version. " + "'published' is the version that was in the file before this run, " + "which is the version a release publishes." + ), + ) args = parser.parse_args() target_path = find_target_pyproject(args.package_name) + with open(target_path, "rb") as f: + published_version = Version(tomllib.load(f)["project"]["version"]) new_version = update_target_version(target_path, args.bump_type) - modified_deps = update_dependents(args.package_name, new_version, target_path) + pinned = new_version if args.dependents == "bumped" else published_version + modified_deps = update_dependents(args.package_name, pinned, target_path) all_modified = [target_path] + modified_deps relative_paths = [str(p.relative_to(REPO_ROOT)) for p in all_modified] diff --git a/scripts/export_release_artifact.py b/scripts/export_release_artifact.py index b9ba8c7f997..0fd30eae299 100644 --- a/scripts/export_release_artifact.py +++ b/scripts/export_release_artifact.py @@ -3,6 +3,9 @@ Run on EVERY release (uv run python scripts/export_release_artifact.py): always writes the package release info; additionally writes the protocol artifact when this release introduces a new protocol version. + +Artifacts are frozen once written. Running this again for the same version +writes nothing and succeeds, so a release can re-run it safely. """ import sys @@ -19,6 +22,14 @@ def main() -> None: # Import the package so every versioned object is registered. import syft_client # noqa: F401 + if client_registry.protocol_bump_missing(): + sys.exit( + "The syft-client protocol changed since the released " + f"protocol-{client_registry.latest_released_protocol_version()}.json; " + "bump SYFT_CLIENT_PROTOCOL_VERSION in " + "syft_client/migrations/registry.py before releasing." + ) + if client_registry.protocol_changed_without_bump(): sys.exit( "The syft-client protocol changed compared to the released " @@ -27,29 +38,21 @@ def main() -> None: "before releasing." ) + PACKAGE_ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True) + PROTOCOLS_DIR.mkdir(parents=True, exist_ok=True) + info_path = PACKAGE_ARTIFACTS_DIR / f"syft-client-{SYFT_CLIENT_VERSION}.json" protocol_path = PROTOCOLS_DIR / f"protocol-{SYFT_CLIENT_PROTOCOL_VERSION}.json" - need_info = not info_path.exists() - need_protocol = not protocol_path.exists() - # Single exit when there is nothing left to write - if not need_info and not need_protocol: - sys.exit( - f"Release artifacts already present:\n" - f" {info_path}\n" - f" {protocol_path}\n" - "They are frozen once written. Bump SYFT_CLIENT_VERSION (and " - "SYFT_CLIENT_PROTOCOL_VERSION if the protocol changed) before " - "exporting again." - ) - - if need_info: + if info_path.exists(): + print(f"Package artifact already present: {info_path}") + else: client_registry.compute_released_package_protocol_info().save(info_path) print(f"Wrote {info_path}") - else: - print(f"Package artifact already present: {info_path}") - if need_protocol: + if protocol_path.exists(): + print(f"Protocol artifact already present: {protocol_path}") + else: client_registry.compute_released_protocol().save(protocol_path) print(f"Wrote {protocol_path} (new protocol version)") diff --git a/syft_client/migrations/registry.py b/syft_client/migrations/registry.py index c071e4c6da8..e74077c855f 100644 --- a/syft_client/migrations/registry.py +++ b/syft_client/migrations/registry.py @@ -14,6 +14,11 @@ # fields on every versioned object. SYFT_CLIENT_PROTOCOL_VERSION = "1" +# Oldest syft-client protocol this release still reads. "0" refuses no peer. +# Raise it only when the code drops support for a released protocol, because a +# peer below the floor cannot exchange syft-client messages with this release. +MIN_SUPPORTED_SYFT_CLIENT_PROTOCOL_VERSION = "0" + # Package-local registry for all versioned syft-client objects. The current # protocol schema is computed from the objects registered into it. client_registry = MigrationRegistry( @@ -21,6 +26,7 @@ package_name=PACKAGE_NAME, package_version=SYFT_CLIENT_VERSION, protocol_version=SYFT_CLIENT_PROTOCOL_VERSION, + min_supported_protocol_version=MIN_SUPPORTED_SYFT_CLIENT_PROTOCOL_VERSION, ) # Shared service for loading/migrating syft-client objects. diff --git a/syft_client/sync/checkpoints/checkpoint.py b/syft_client/sync/checkpoints/checkpoint.py index b2a1ae41c60..26d71711fd5 100644 --- a/syft_client/sync/checkpoints/checkpoint.py +++ b/syft_client/sync/checkpoints/checkpoint.py @@ -11,12 +11,14 @@ - After N incremental checkpoints: compact into single full Checkpoint """ -from typing import List, Dict, TYPE_CHECKING -from pydantic import BaseModel, Field from pathlib import Path +from typing import TYPE_CHECKING, Dict, List + +from pydantic import BaseModel, Field + from syft_client.sync.utils.syftbox_utils import ( - create_event_timestamp, compress_data, + create_event_timestamp, uncompress_data, ) @@ -28,6 +30,21 @@ INCREMENTAL_CHECKPOINT_PREFIX = "incremental_checkpoint" CHECKPOINT_VERSION = 1 + +def _check_version(version: int, kind: str) -> None: + """Refuse a checkpoint from a later client.""" + + # A later client can change what a field holds while the object still parses. + # The restore would then be wrong and silent. Every caller falls back to a + # download of all events, so a refusal costs one slow cold start. + + if version > CHECKPOINT_VERSION: + raise ValueError( + f"This {kind} has version {version}, and this client reads up to " + f"version {CHECKPOINT_VERSION}." + ) + + # Default compacting threshold: merge after this many incremental checkpoints DEFAULT_COMPACTING_THRESHOLD = 4 @@ -124,7 +141,9 @@ def as_compressed_data(self) -> bytes: def from_compressed_data(cls, data: bytes) -> "Checkpoint": """Load checkpoint from compressed data.""" uncompressed_data = uncompress_data(data) - return cls.model_validate_json(uncompressed_data) + checkpoint = cls.model_validate_json(uncompressed_data) + _check_version(checkpoint.version, "checkpoint") + return checkpoint class IncrementalCheckpoint(BaseModel): @@ -180,7 +199,9 @@ def as_compressed_data(self) -> bytes: def from_compressed_data(cls, data: bytes) -> "IncrementalCheckpoint": """Load from compressed data.""" uncompressed_data = uncompress_data(data) - return cls.model_validate_json(uncompressed_data) + checkpoint = cls.model_validate_json(uncompressed_data) + _check_version(checkpoint.version, "incremental checkpoint") + return checkpoint def compact_incremental_checkpoints( diff --git a/syft_client/sync/checkpoints/rolling_state.py b/syft_client/sync/checkpoints/rolling_state.py index cbbf6824895..31be37e60da 100644 --- a/syft_client/sync/checkpoints/rolling_state.py +++ b/syft_client/sync/checkpoints/rolling_state.py @@ -13,22 +13,36 @@ """ from typing import List + from pydantic import BaseModel, Field -from syft_client.sync.utils.syftbox_utils import ( - create_event_timestamp, - compress_data, - uncompress_data, -) + from syft_client.sync.events.file_change_event import ( FileChangeEvent, FileChangeEventsMessage, ) - +from syft_client.sync.utils.syftbox_utils import ( + compress_data, + create_event_timestamp, + uncompress_data, +) ROLLING_STATE_FILENAME_PREFIX = "rolling_state" ROLLING_STATE_VERSION = 1 +def raise_for_later_version(version: int) -> None: + """Refuse a rolling state from a later client.""" + + # A later client can change what a field holds while the object still + # parses. The restore would then be wrong and silent. Every caller falls + # back to a download of all events, so a refusal costs one slow cold start. + if version > ROLLING_STATE_VERSION: + raise ValueError( + f"This rolling state has version {version}, and this client reads up " + f"to version {ROLLING_STATE_VERSION}." + ) + + class RollingState(BaseModel): """ Rolling state keeps the latest state of each file since the last checkpoint. @@ -111,7 +125,9 @@ def as_compressed_data(self) -> bytes: def from_compressed_data(cls, data: bytes) -> "RollingState": """Load rolling state from compressed data.""" uncompressed_data = uncompress_data(data) - return cls.model_validate_json(uncompressed_data) + state = cls.model_validate_json(uncompressed_data) + raise_for_later_version(state.version) + return state @classmethod def filename_to_timestamp(cls, filename: str) -> float | None: diff --git a/syft_client/sync/connections/connection_router.py b/syft_client/sync/connections/connection_router.py index e165ed86e21..38c18eb5301 100644 --- a/syft_client/sync/connections/connection_router.py +++ b/syft_client/sync/connections/connection_router.py @@ -1,29 +1,38 @@ -from pydantic import BaseModel +import logging from typing import TYPE_CHECKING, List, Optional + +from pydantic import BaseModel + +from syft_client.sync.checkpoints.checkpoint import Checkpoint, IncrementalCheckpoint +from syft_client.sync.checkpoints.rolling_state import RollingState from syft_client.sync.connections.base_connection import ( ConnectionConfig, FileCollection, SyftboxPlatformConnection, ) -from syft_client.sync.connections.drive.gdrive_transport import GDriveConnection +from syft_client.sync.connections.drive.gdrive_transport import ( + PEERS_META_KEY, + GDriveConnection, +) from syft_client.sync.events.file_change_event import ( FileChangeEventsMessage, ) -from syft_client.sync.checkpoints.checkpoint import Checkpoint, IncrementalCheckpoint -from syft_client.sync.checkpoints.rolling_state import RollingState -from syft_client.sync.peers.peer_store import PeerStore from syft_client.sync.messages.proposed_filechange import ProposedFileChangesMessage -from syft_client.sync.platforms.gdrive_files_platform import GdriveFilesPlatform from syft_client.sync.peers.peer import Peer, PeerState +from syft_client.sync.peers.peer_store import PeerStore +from syft_client.sync.platforms.gdrive_files_platform import GdriveFilesPlatform from syft_client.sync.utils.print_utils import ( - print_peer_adding_to_platform, print_peer_added_to_platform, + print_peer_adding_to_platform, ) if TYPE_CHECKING: from syft_client.sync.version.version_info import VersionInfo +logger = logging.getLogger(__name__) + + class ConnectionRouter(BaseModel): connections: List[SyftboxPlatformConnection] @@ -194,9 +203,19 @@ def get_all_peers_from_json(self, force_download: bool = False) -> List[Peer]: peers_data = connection._get_peers_json(force_download=force_download) peers = [] for email, data in peers_data.items(): + if email == PEERS_META_KEY: + continue try: state = PeerState(data.get("state", "unknown")) except ValueError: + # A later client wrote a state that this client does not know. + # The writer changes one entry and keeps the rest, so the entry + # stays in the file. The peer returns after an upgrade. + logger.warning( + f"Skipping peer {email}: unknown state " + f"{data.get('state')!r}. Install a newer syft-client to see " + "this peer." + ) continue peer = Peer( email=email, diff --git a/syft_client/sync/connections/drive/gdrive_transport.py b/syft_client/sync/connections/drive/gdrive_transport.py index 3cf50f2dd7c..e5476adb025 100644 --- a/syft_client/sync/connections/drive/gdrive_transport.py +++ b/syft_client/sync/connections/drive/gdrive_transport.py @@ -1,58 +1,58 @@ """Google Drive Files transport layer implementation""" -import logging import io import json -from pathlib import Path +import logging import pickle -from syft_client.sync.utils.syftbox_utils import check_env -from syft_client.version import SYFT_CLIENT_VERSION -from typing import Any, Dict, List, Optional, Tuple -from typing import TYPE_CHECKING -from pydantic import BaseModel +from pathlib import Path +from typing import TYPE_CHECKING, Any, Optional + +from google.oauth2.credentials import Credentials as GoogleCredentials from google_auth_httplib2 import AuthorizedHttp from googleapiclient.discovery import build from googleapiclient.http import MediaIoBaseDownload, MediaIoBaseUpload, build_http -from google.oauth2.credentials import Credentials as GoogleCredentials +from pydantic import BaseModel +from syft_datasets.dataset_manager import ( + DATASET_COLLECTION_PREFIX, + PRIVATE_DATASET_COLLECTION_PREFIX, +) +from syft_migration import MigrationError -from syft_client.sync.connections.drive.gdrive_utils import ( - gather_all_file_and_folder_ids_recursive, +from syft_client.sync.checkpoints.checkpoint import ( + CHECKPOINT_FILENAME_PREFIX, + INCREMENTAL_CHECKPOINT_PREFIX, + Checkpoint, + IncrementalCheckpoint, ) -from syft_client.sync.connections.drive.gdrive_retry import ( - execute_with_retries, - next_chunk_with_retries, - batch_execute_with_retries, +from syft_client.sync.checkpoints.rolling_state import ( + ROLLING_STATE_FILENAME_PREFIX, + RollingState, ) -from syft_client.sync.version.version_info import _parse_semver - from syft_client.sync.connections.base_connection import ( FileCollection, SyftboxPlatformConnection, ) -from syft_datasets.dataset_manager import ( - DATASET_COLLECTION_PREFIX, - PRIVATE_DATASET_COLLECTION_PREFIX, +from syft_client.sync.connections.drive.gdrive_retry import ( + batch_execute_with_retries, + execute_with_retries, + next_chunk_with_retries, ) +from syft_client.sync.connections.drive.gdrive_utils import ( + gather_all_file_and_folder_ids_recursive, +) +from syft_client.sync.environments.environment import Environment from syft_client.sync.events.file_change_event import ( - FileChangeEventsMessageFileName, FileChangeEventsMessage, + FileChangeEventsMessageFileName, ) from syft_client.sync.messages.proposed_filechange import ( - MessageFileName, FileNameParseError, + MessageFileName, ProposedFileChangesMessage, ) -from syft_client.sync.environments.environment import Environment -from syft_client.sync.checkpoints.checkpoint import ( - Checkpoint, - IncrementalCheckpoint, - CHECKPOINT_FILENAME_PREFIX, - INCREMENTAL_CHECKPOINT_PREFIX, -) -from syft_client.sync.checkpoints.rolling_state import ( - RollingState, - ROLLING_STATE_FILENAME_PREFIX, -) +from syft_client.sync.utils.syftbox_utils import check_env +from syft_client.sync.version.version_info import _parse_semver +from syft_client.version import SYFT_CLIENT_VERSION if TYPE_CHECKING: from syft_client.sync.connections.drive.grdrive_config import ( @@ -80,8 +80,8 @@ def build_drive_service( http = build_http() http.timeout = timeout if environment == Environment.COLAB: - from google.colab import auth as colab_auth import google.auth + from google.colab import auth as colab_auth colab_auth.authenticate_user() creds, _ = google.auth.default() @@ -96,6 +96,15 @@ def build_drive_service( LEGACY_GDRIVE_OUTBOX_INBOX_FOLDER_PREFIX = "syft_outbox_inbox" # legacy prefix GDRIVE_P2P_FOLDER_DATASITE_PREFIX = "syft_datasite" SYFT_PEERS_FILE = "SYFT_peers.json" + +# SYFT_peers.json is a flat map of peer email to entry, so a version at the top +# level would look like a peer email. The version goes under this reserved key. +# A client written before the key reads a peer state from that entry and fails. +# The key therefore never appears as a peer. +PEERS_META_KEY = "_meta" +# Shape of one entry in SYFT_peers.json. Raise this when an entry changes. A file +# with no reserved entry was written before the version, and is version 0. +SYFT_PEERS_VERSION = 1 SYFT_VERSION_FILE = "SYFT_version.json" @@ -221,33 +230,70 @@ def _extract_version_from_name(name: str) -> str | None: return None -def _filter_patch_compatible( +# A folder id and name, with the version from the name. The version fields come +# first, so the default sort puts these in version order. +_VersionedFolder = tuple[int, int, int, str, str] + + +def _sorted_by_version(folders: list[tuple[str, str]]) -> list[tuple[str, str]]: + """Folders from the lowest version to the highest. + + A name with no readable version sorts first, so a versioned folder always + wins when the caller takes the last entry. + """ + + def key(entry: tuple[str, str]) -> tuple[int, int, int]: + version_str = _extract_version_from_name(entry[1]) + if version_str is None: + return (-1, -1, -1) + try: + return _parse_semver(version_str) + except ValueError: + return (-1, -1, -1) + + return sorted(folders, key=key) + + +def _partition_by_version( folders: list[tuple[str, str]], current_version: str | None = None, -) -> list[tuple[str, str]]: - """Keep folders whose embedded version has matching major.minor. +) -> tuple[list[tuple[str, str]], list[tuple[str, str]], list[tuple[str, str]]]: + """Split folders into (compatible, older, newer) by the version in the name. - `current_version` defaults to the module-level SYFT_CLIENT_VERSION at call - time (not import time) so tests that patch the version take effect. + Compatible means the same major and minor as the current version. The function + drops a folder that has no version in its name. Each list starts at the lowest + version. """ if current_version is None: current_version = SYFT_CLIENT_VERSION try: - cur_major, cur_minor, _ = _parse_semver(current_version) + current = _parse_semver(current_version) except ValueError: - return [] - kept: list[tuple[str, str]] = [] + return [], [], [] + + compatible: list[_VersionedFolder] = [] + older: list[_VersionedFolder] = [] + newer: list[_VersionedFolder] = [] for fid, name in folders: version_str = _extract_version_from_name(name) if version_str is None: continue try: - major, minor, _ = _parse_semver(version_str) + found = _parse_semver(version_str) except ValueError: continue - if major == cur_major and minor == cur_minor: - kept.append((fid, name)) - return kept + entry = (*found, fid, name) + if found[:2] == current[:2]: + compatible.append(entry) + elif found < current: + older.append(entry) + else: + newer.append(entry) + + def _ordered(entries: list[_VersionedFolder]) -> list[tuple[str, str]]: + return [(fid, name) for *_, fid, name in sorted(entries)] + + return _ordered(compatible), _ordered(older), _ordered(newer) class GDriveConnection(SyftboxPlatformConnection): @@ -272,21 +318,21 @@ class Config: _personal_syftbox_folder_id: str | None = None # peer_email -> folder_id (folders I created for peer's datasite) - peer_datasite_inbox_cache: Dict[str, str] = {} - peer_datasite_outbox_cache: Dict[str, str] = {} + peer_datasite_inbox_cache: dict[str, str] = {} + peer_datasite_outbox_cache: dict[str, str] = {} # peer_email -> folder_id (folders peer created for my datasite) - own_datasite_inbox_cache: Dict[str, str] = {} - own_datasite_outbox_cache: Dict[str, str] = {} + own_datasite_inbox_cache: dict[str, str] = {} + own_datasite_outbox_cache: dict[str, str] = {} # sender email -> archive folder id - archive_folder_id_cache: Dict[str, str] = {} + archive_folder_id_cache: dict[str, str] = {} # fname -> gdrive id - personal_syftbox_event_id_cache: Dict[str, str] = {} + personal_syftbox_event_id_cache: dict[str, str] = {} # tag -> dataset collection folder id - dataset_collection_folder_id_cache: Dict[str, str] = {} + dataset_collection_folder_id_cache: dict[str, str] = {} # Rolling state caches for single-API-call optimization _rolling_state_folder_id: str | None = None @@ -296,7 +342,7 @@ class Config: _encryption_bundles_folder_id: str | None = None # Cached SYFT_peers.json contents (None = not loaded yet). - _peers_json_cache: Dict[str, Dict[str, str]] | None = None + _peers_json_cache: dict[str, dict[str, str]] | None = None @classmethod def from_config(cls, config: "GdriveConnectionConfig") -> "GDriveConnection": @@ -470,7 +516,7 @@ def _get_peers_file_id(self) -> str | None: items = results.get("files", []) return items[0]["id"] if items else None - def _download_peers_json(self) -> Dict[str, Dict[str, str]]: + def _download_peers_json(self) -> dict[str, dict[str, str]]: """Fetch peers JSON from GDrive. Returns empty dict if not found.""" file_id = self._get_peers_file_id() if file_id is None: @@ -478,22 +524,30 @@ def _download_peers_json(self) -> Dict[str, Dict[str, str]]: try: file_data = self.download_file(file_id) - return json.loads(file_data.decode("utf-8")) except Exception as e: - print(f"Warning: Error reading peers file: {e}") + print(f"Warning: could not download the peers file: {e}") + return {} + try: + return json.loads(file_data.decode("utf-8")) + except ValueError as e: + print(f"Warning: could not read the peers file: {e}") return {} def _get_peers_json( self, force_download: bool = False - ) -> Dict[str, Dict[str, str]]: + ) -> dict[str, dict[str, str]]: """Return peers JSON, using the in-memory cache when available.""" if self._peers_json_cache is not None and not force_download: return self._peers_json_cache self._peers_json_cache = self._download_peers_json() return self._peers_json_cache - def _write_peers_json(self, peers_data: Dict[str, Dict[str, str]]): + def _write_peers_json(self, peers_data: dict[str, dict[str, str]]): """Write peers JSON to GDrive. Creates or updates the file.""" + peers_data = { + **peers_data, + PEERS_META_KEY: {"version": SYFT_PEERS_VERSION}, + } syftbox_folder_id = self.get_syftbox_folder_id() file_id = self._get_peers_file_id() @@ -542,7 +596,7 @@ def _update_peer_state( peers_data[peer_email] = existing self._write_peers_json(peers_data) - def get_peer_requests(self) -> List[str]: + def get_peer_requests(self) -> list[str]: """Get list of pending peer requests. Scans for syft_datasite_#version#{self}_* folders NOT owned by self — those are @@ -563,10 +617,12 @@ def get_peer_requests(self) -> List[str]: for f in results.get("files", []): try: folder = GdriveP2PFolder.from_name(f["name"]) - if folder.datasite_email == self.email: - all_folder_peers.add(folder.peer_email) - except (ValueError, Exception): + except ValueError: + # The query matches a name prefix, so a folder with another shape + # can appear here. continue + if folder.datasite_email == self.email: + all_folder_peers.add(folder.peer_email) peers_data = self._get_peers_json() pending_peers = [] @@ -609,7 +665,7 @@ def watcher_download_raw_events_from_outbox( def watcher_get_events_messages( self, peer_email: str, since_timestamp: float | None - ) -> List[FileChangeEventsMessage]: + ) -> list[FileChangeEventsMessage]: raw_list = self.watcher_download_raw_events_from_outbox( peer_email, since_timestamp ) @@ -617,7 +673,7 @@ def watcher_get_events_messages( def watcher_get_outbox_file_metadatas( self, peer_email: str, since_timestamp: float | None - ) -> List[Dict]: + ) -> list[dict]: """Get file metadata from peer's outbox folder without downloading.""" folder_id = self._get_peer_datasite_outbox_id(peer_email) if folder_id is None: @@ -667,7 +723,7 @@ def owner_download_raw_bytes_by_id(self, file_id: str) -> bytes: def owner_get_all_accepted_event_file_ids( self, since_timestamp: float | None = None - ) -> List[str]: + ) -> list[str]: personal_syftbox_folder_id = self.get_personal_syftbox_folder_id() file_metadatas = self.get_file_metadatas_from_folder( personal_syftbox_folder_id, since_timestamp=since_timestamp @@ -689,7 +745,7 @@ def owner_download_all_raw_events_from_syftbox(self) -> list[bytes]: try: file_data = self.download_file(gdrive_id) except Exception as e: - print(e) + print(f"Warning: could not download event {fname_obj.as_string()}: {e}") continue result.append(file_data) return result @@ -822,7 +878,10 @@ def get_personal_syftbox_folder_id(self) -> str: # '#{peer}#{type}#{email}'. Personal folder shape is exactly # '{version}#{email}', so require a single '#'. folders = [(fid, name) for fid, name in folders if name.count("#") == 1] - folder_id = self._expect_one(_filter_patch_compatible(folders)) + folder_id = self._find_or_adopt_versioned_folder( + folders, + current_name=GdrivePersonalSyftboxFolder(email=self.email).as_string(), + ) if folder_id: self._personal_syftbox_folder_id = folder_id return folder_id @@ -901,7 +960,7 @@ def get_file_metadatas_from_folder( folder_id: str, since_timestamp: float | None = None, page_size: int = 100, - ) -> List[Dict]: + ) -> list[dict]: """ Get file metadatas from folder with early termination. @@ -966,37 +1025,39 @@ def get_file_metadatas_from_folder( @staticmethod def _filter_valid_file_metadatas( - file_metadatas: List[Dict], - ) -> List[Dict]: + file_metadatas: list[dict], + ) -> list[dict]: res = [] for file_metadata in file_metadatas: fname = file_metadata["name"] try: - _ = FileChangeEventsMessageFileName.from_string(fname) - res.append(file_metadata) - except Exception: + FileChangeEventsMessageFileName.from_string(fname) + except ValueError: + # The folder holds other files, so a name that is not an event + # name is normal here. This method filters them out. continue + res.append(file_metadata) return res @staticmethod def _get_valid_events_from_file_metadatas( - file_metadatas: List[Dict], - ) -> List[FileChangeEventsMessageFileName]: + file_metadatas: list[dict], + ) -> list[FileChangeEventsMessageFileName]: res = [] for file_metadata in file_metadatas: fname = file_metadata["name"] try: message_filename = FileChangeEventsMessageFileName.from_string(fname) - res.append(message_filename) - except Exception: - print("Warning, invalid file name: ", fname) + except ValueError: + print(f"Warning: invalid event file name: {fname}") continue + res.append(message_filename) return res @staticmethod def _get_valid_messages_from_file_metadatas( - file_metadatas: List[Dict], - ) -> List[MessageFileName]: + file_metadatas: list[dict], + ) -> list[MessageFileName]: res = [] for file_metadata in file_metadatas: try: @@ -1065,8 +1126,21 @@ def _is_exact_match(name: str) -> bool: and folder.peer_email == peer_email ) - folders = [(fid, name) for fid, name in folders if _is_exact_match(name)] - return self._expect_one(_filter_patch_compatible(folders)) + # Ignore the version in the name. Each peer builds this name from its own + # client version, so a filter here hides the folder that the peer uses. + # After an upgrade the client therefore finds the old folder and writes to + # it. It makes no second folder, which an older peer would never look for. + candidates = _sorted_by_version( + [(fid, name) for fid, name in folders if _is_exact_match(name)] + ) + if not candidates: + return None + if len(candidates) > 1: + print( + f"Warning: {len(candidates)} P2P folders for {datasite_email} " + f"{folder_type} {peer_email}; using {candidates[-1][1]}" + ) + return candidates[-1][0] def _get_peer_datasite_inbox_id(self, peer_email: str) -> str | None: """Get folder: syft_datasite_{peer}_inbox_{self}, owned by self.""" @@ -1180,7 +1254,7 @@ def reset_caches(self): self._encryption_bundles_folder_id = None self._peers_json_cache = None - def gather_all_file_and_folder_ids(self) -> List[str]: + def gather_all_file_and_folder_ids(self) -> list[str]: syftbox_folder_id = self.get_syftbox_folder_id() return gather_all_file_and_folder_ids_recursive( self.drive_service, syftbox_folder_id @@ -1188,7 +1262,7 @@ def gather_all_file_and_folder_ids(self) -> List[str]: def delete_multiple_files_by_ids( self, - file_ids: List[str], + file_ids: list[str], ignore_permissions_errors: bool = True, ignore_file_not_found: bool = True, ): @@ -1226,17 +1300,13 @@ def callback(request_id, response, exception): batch.add(self.drive_service.files().delete(fileId=file_id)) batch_execute_with_retries(batch) - def delete_file_by_id( - self, file_id: str, verbose: bool = False, raise_on_error: bool = False - ): + def delete_file_by_id(self, file_id: str, raise_on_error: bool = False): try: execute_with_retries(self.drive_service.files().delete(fileId=file_id)) except Exception as e: if raise_on_error: raise e - else: - if verbose: - print(f"Error deleting file: {file_id}") + print(f"Warning: could not delete file {file_id}: {e}") def delete_unversioned_state(self) -> None: """Delete non-versioned remote artifacts during upgrade. @@ -1350,7 +1420,7 @@ def find_orphaned_message_files(self) -> list[str]: return file_ids - def create_file_payload(self, data: Any) -> Tuple[MediaIoBaseUpload, str]: + def create_file_payload(self, data: Any) -> tuple[MediaIoBaseUpload, str]: """Create a file payload for the GDrive""" if isinstance(data, str): file_data = data.encode("utf-8") @@ -1402,7 +1472,8 @@ def _find_folders( Thin wrapper over Drive's files.list -- handles query building and pagination, knows nothing about versions. Pair with - _filter_patch_compatible when the caller cares about version compat. + _partition_by_version or _sorted_by_version when the caller cares about + the version in the folder name. """ clauses = [f"mimeType='{GOOGLE_FOLDER_MIME_TYPE}'", "trashed=false"] for substr in name_contains: @@ -1447,6 +1518,54 @@ def _expect_one(self, folders: list[tuple[str, str]]) -> str | None: f"folder(s) on Drive (keeping the one with your data) and retry." ) + def _find_or_adopt_versioned_folder( + self, + folders: list[tuple[str, str]], + current_name: str, + current_version: str | None = None, + ) -> str | None: + """Return the id of a PRIVATE folder for this client version, or None. + + A private folder name holds the client version, so a minor upgrade looks + for a name that does not exist yet. This method renames the folder of the + highest earlier version to `current_name` and keeps the data. A new folder + would leave the data of the user on Drive and out of reach. + + Renames the folder, so the caller must own it and no peer may look it up by + name. A P2P folder fails both conditions: use `_expect_one` for those. + + Raises RuntimeError if only a folder from a later version exists, or if + more than one compatible folder exists. + """ + compatible, older, newer = _partition_by_version(folders, current_version) + if compatible: + return self._expect_one(compatible) + if newer: + names = [n for _, n in newer] + latest = _extract_version_from_name(names[-1]) + raise RuntimeError( + f"Found a folder from a later client version on Drive: {names}. " + f"This client is {current_version or SYFT_CLIENT_VERSION} and " + f"cannot read that data. Install syft-client {latest} or later." + ) + if not older: + return None + + folder_id, name = older[-1] + execute_with_retries( + self.drive_service.files().update( + fileId=folder_id, body={"name": current_name} + ) + ) + print(f"Adopted the folder of an earlier version: {name} -> {current_name}") + if len(older) > 1: + stale = [n for _, n in older[:-1]] + print( + f"Warning: {len(stale)} folder(s) of earlier versions stay on " + f"Drive: {stale}" + ) + return folder_id + def download_file(self, file_id: str) -> bytes: request = self.drive_service.files().get_media(fileId=file_id) @@ -1457,7 +1576,7 @@ def download_file(self, file_id: str) -> bytes: done = False while not done: - status, done = next_chunk_with_retries(downloader) + _, done = next_chunk_with_retries(downloader) message_data = file_buffer.getvalue() return message_data @@ -1541,10 +1660,9 @@ def _batch_add_permissions(self, file_id: str, users: list[str]) -> None: """Add reader permissions for multiple users in a single batch request.""" def callback(request_id, response, exception): - if exception: - # Ignore "already shared" errors - if "alreadyShared" not in str(exception): - raise exception + # Ignore "already shared" errors + if exception and "alreadyShared" not in str(exception): + raise exception BATCH_SIZE = 100 for i in range(0, len(users), BATCH_SIZE): @@ -1620,23 +1738,21 @@ def owner_list_all_dataset_collections_with_permissions( collections = [] for folder in results.get("files", []): - folder_id = folder["id"] try: folder_obj = DatasetCollectionFolder.from_name(folder["name"]) - has_anyone = ( - folder.get("appProperties", {}).get("syft_shared_with_any") - == "true" - ) - collections.append( - FileCollection( - folder_id=folder_id, - tag=folder_obj.tag, - content_hash=folder_obj.content_hash, - has_any_permission=has_anyone, - ) - ) - except Exception: + except ValueError: continue + has_anyone = ( + folder.get("appProperties", {}).get("syft_shared_with_any") == "true" + ) + collections.append( + FileCollection( + folder_id=folder["id"], + tag=folder_obj.tag, + content_hash=folder_obj.content_hash, + has_any_permission=has_anyone, + ) + ) return collections @@ -1705,7 +1821,7 @@ def watcher_download_dataset_collection( def watcher_get_dataset_collection_file_metadatas( self, tag: str, content_hash: str, owner_email: str - ) -> List[Dict]: + ) -> list[dict]: """Get file metadata from a dataset collection without downloading.""" folder_obj = DatasetCollectionFolder(tag=tag, content_hash=content_hash) folder_name = folder_obj.as_string() @@ -1822,7 +1938,7 @@ def owner_delete_private_dataset_collection(self, tag: str) -> None: def owner_get_private_collection_file_metadatas( self, tag: str, content_hash: str, owner_email: str - ) -> List[Dict]: + ) -> list[dict]: """Get file metadata from a private dataset collection without downloading.""" folder_obj = PrivateDatasetCollectionFolder(tag=tag, content_hash=content_hash) folder_name = folder_obj.as_string() @@ -1916,8 +2032,13 @@ def read_own_version_file(self) -> Optional["VersionInfo"]: try: file_data = self.download_file(file_id) + except Exception as e: + print(f"Warning: could not download the own version file: {e}") + return None + try: return VersionInfo.from_json(file_data.decode("utf-8")) - except Exception: + except (ValueError, MigrationError) as e: + print(f"Warning: could not read the own version file: {e}") return None def read_peer_version_file(self, peer_email: str) -> Optional["VersionInfo"]: @@ -1930,8 +2051,13 @@ def read_peer_version_file(self, peer_email: str) -> Optional["VersionInfo"]: try: file_data = self.download_file(file_id) + except Exception as e: + print(f"Warning: could not download the version file of {peer_email}: {e}") + return None + try: return VersionInfo.from_json(file_data.decode("utf-8")) - except Exception: + except (ValueError, MigrationError) as e: + print(f"Warning: could not read the version file of {peer_email}: {e}") return None def share_version_file_with_peer(self, peer_email: str) -> None: @@ -1961,7 +2087,9 @@ def _get_checkpoints_folder_id(self) -> str | None: name_contains=[f"{self.email}-", "-checkpoints"], parent_id=self.get_syftbox_folder_id(), ) - return self._expect_one(_filter_patch_compatible(folders)) + return self._find_or_adopt_versioned_folder( + folders, current_name=self._get_checkpoints_folder_name() + ) def _get_or_create_checkpoints_folder_id(self) -> str: """Get or create the checkpoints folder.""" @@ -2264,7 +2392,9 @@ def _get_rolling_state_folder_id(self, use_cache: bool = True) -> str | None: name_contains=[f"{self.email}-", "-rolling-state"], parent_id=self.get_syftbox_folder_id(), ) - folder_id = self._expect_one(_filter_patch_compatible(folders)) + folder_id = self._find_or_adopt_versioned_folder( + folders, current_name=self._get_rolling_state_folder_name() + ) if folder_id is not None: self._rolling_state_folder_id = folder_id return folder_id @@ -2296,7 +2426,13 @@ def upload_raw_rolling_state(self, filename: str, data: bytes) -> str: media_body=payload, ).execute() return self._rolling_state_file_id - except Exception: + except Exception as e: + # The cached file is gone or unreachable. Clear the cache and + # write a new file below. + print( + f"Warning: could not update rolling state " + f"{self._rolling_state_file_id}, writing a new file: {e}" + ) self._rolling_state_file_id = None folder_id = self._get_or_create_rolling_state_folder_id() @@ -2451,6 +2587,11 @@ def read_peer_encryption_bundle(self, peer_email: str) -> str | None: return None try: data = self.download_file(items[0]["id"]) + except Exception as e: + print(f"Warning: could not download the bundle of {peer_email}: {e}") + return None + try: return data.decode("utf-8") - except Exception: + except ValueError as e: + print(f"Warning: could not read the bundle of {peer_email}: {e}") return None diff --git a/syft_client/sync/peers/peer_store.py b/syft_client/sync/peers/peer_store.py index e95a1d7dc96..5c1c662795d 100644 --- a/syft_client/sync/peers/peer_store.py +++ b/syft_client/sync/peers/peer_store.py @@ -20,6 +20,11 @@ PRIVATE_DIR_NAME = "private" CRYPTO_KEYS_FILENAME = "crypto_keys.json" +# Format of the crypto key file. Raise it when the layout of the file changes, +# and add a read path for every earlier version. A file with no version was +# written before the field, and is version 0. +CRYPTO_KEYS_VERSION = 1 + def datasite_crypto_keys_path(syftbox_folder: Path | str, email: str) -> Path: """Per-datasite key file: ``//private/crypto_keys.json``.""" @@ -230,6 +235,7 @@ def decrypt_and_verify_for_self_if_needed(self, data: bytes) -> bytes: def save_keys(self, path: Path) -> None: keys = self._ensure_private_keys() data = { + "version": CRYPTO_KEYS_VERSION, "email": self.email, "keys_jwk": keys.to_jwks(), "peer_bundles": { @@ -245,6 +251,16 @@ def save_keys(self, path: Path) -> None: @classmethod def load_keys(cls, path: Path) -> "PeerStore": data = json.loads(Path(path).read_text()) + # A file with no version was written before the field, and its layout is + # this client reads. A later version is refused: a user cannot rebuild a + # private key, so a wrong read loses the keys. + version = data.get("version", 0) + if version > CRYPTO_KEYS_VERSION: + raise ValueError( + f"The crypto key file at {path} has version {version}, and this " + f"client reads up to version {CRYPTO_KEYS_VERSION}. Install a " + "newer syft-client to use these keys." + ) store = cls(email=data["email"], use_encryption=True) store._private_keys = syc.SyftPrivateKeys.from_jwks(data["keys_jwk"]) for email, bundle_dict in data.get("peer_bundles", {}).items(): diff --git a/syft_client/sync/sync/caches/persisted_dict.py b/syft_client/sync/sync/caches/persisted_dict.py index 24064b4ce99..5e43ee0fe9c 100644 --- a/syft_client/sync/sync/caches/persisted_dict.py +++ b/syft_client/sync/sync/caches/persisted_dict.py @@ -31,6 +31,11 @@ import portalocker +# Format of the persisted file: {"version": N, "entries": {...}}. Raise it when +# the layout of an entry changes. A file with no version holds the entries at the +# top level, was written before the field, and is version 0. +PERSISTED_DICT_VERSION = 1 + class PersistedDict(dict): """Dict that persists to a JSON file. With path=None it's a plain in-memory dict.""" @@ -94,10 +99,28 @@ def _read_from_file(self) -> None: return try: data = json.loads(self._path.read_text()) - for k, v in data.items(): - super().__setitem__(self._key_deserializer(k), v) except (json.JSONDecodeError, OSError): - pass + return + entries = self._entries_of(data) + for k, v in entries.items(): + super().__setitem__(self._key_deserializer(k), v) + + @staticmethod + def _entries_of(data: Any) -> dict: + """The entries to load from a parsed file, empty when it cannot be read. + + The client rebuilds every cache that uses this class, so an unreadable + file costs a re-scan and nothing else. A file from a later version + therefore starts empty instead of stopping the client. + """ + if not isinstance(data, dict): + return {} + if "version" not in data or "entries" not in data: + # Written before the version field existed: entries at the top level. + return data + if data["version"] > PERSISTED_DICT_VERSION: + return {} + return data["entries"] def _write_to_file(self) -> None: if self._path is None: @@ -106,7 +129,10 @@ def _write_to_file(self) -> None: # Per-process unique tmp path: even with the file lock, this guards # against any path where two writers share a tmp filename. tmp = self._path.with_suffix(f".tmp.{os.getpid()}.{uuid4().hex}") - serialized = {self._key_serializer(k): v for k, v in super().items()} + serialized = { + "version": PERSISTED_DICT_VERSION, + "entries": {self._key_serializer(k): v for k, v in super().items()}, + } try: tmp.write_text(json.dumps(serialized)) tmp.replace(self._path) diff --git a/syft_client/sync/sync/datasite_owner_syncer.py b/syft_client/sync/sync/datasite_owner_syncer.py index 81e6b25241b..538af2f79c5 100644 --- a/syft_client/sync/sync/datasite_owner_syncer.py +++ b/syft_client/sync/sync/datasite_owner_syncer.py @@ -1,38 +1,42 @@ import logging -from pathlib import Path -from uuid import uuid4 - -from pydantic import ConfigDict, Field, BaseModel, PrivateAttr from concurrent.futures import ThreadPoolExecutor +from pathlib import Path from queue import Queue from typing import List, Tuple -from syft_client.sync.events.file_change_event import ( - FileChangeEventsMessage, - FileChangeEventsMessageFileName, - FileChangeEvent, +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, PrivateAttr +from syft_perms import SyftPermContext + +from syft_client.sync.callback_mixin import BaseModelCallbackMixin +from syft_client.sync.checkpoints.checkpoint import ( + DEFAULT_COMPACTING_THRESHOLD, + Checkpoint, + CheckpointFile, + IncrementalCheckpoint, + compact_incremental_checkpoints, +) +from syft_client.sync.checkpoints.rolling_state import ( + RollingState, + raise_for_later_version, ) from syft_client.sync.connections.base_connection import ( ConnectionConfig, FileCollection, ) -from syft_client.sync.sync.caches.datasite_owner_cache import ( - DataSiteOwnerEventCacheConfig, -) from syft_client.sync.connections.connection_router import ConnectionRouter -from syft_client.sync.sync.caches.datasite_owner_cache import DataSiteOwnerEventCache -from syft_client.sync.callback_mixin import BaseModelCallbackMixin +from syft_client.sync.events.file_change_event import ( + FileChangeEvent, + FileChangeEventsMessage, + FileChangeEventsMessageFileName, +) from syft_client.sync.messages.proposed_filechange import ProposedFileChangesMessage -from syft_client.sync.utils.path_filters import is_normal_syncable_path -from syft_client.sync.checkpoints.checkpoint import ( - Checkpoint, - CheckpointFile, - IncrementalCheckpoint, - compact_incremental_checkpoints, - DEFAULT_COMPACTING_THRESHOLD, +from syft_client.sync.sync.caches.datasite_owner_cache import ( + DataSiteOwnerEventCache, + DataSiteOwnerEventCacheConfig, ) -from syft_client.sync.checkpoints.rolling_state import RollingState -from syft_perms import SyftPermContext from syft_client.sync.sync.constants import CACHE_DIR, ROLLING_STATE_FILENAME +from syft_client.sync.utils.path_filters import is_normal_syncable_path logger = logging.getLogger(__name__) @@ -124,9 +128,14 @@ def _load_rolling_state(self) -> None: if not path.exists(): return try: - self._rolling_state = RollingState.model_validate_json(path.read_text()) - except Exception: - pass + state = RollingState.model_validate_json(path.read_text()) + raise_for_later_version(state.version) + except (ValueError, OSError) as e: + # A later client wrote this file, or it is damaged. The caller falls + # back to a download of all events. + print(f"Warning: could not load the local rolling state: {e}") + return + self._rolling_state = state def _save_rolling_state(self) -> None: """Save rolling state to disk for cross-process consistency.""" @@ -537,8 +546,8 @@ def _create_resend_event(self, path: str) -> "FileChangeEvent | None": if content is None: return None from syft_client.sync.utils.syftbox_utils import ( - get_event_hash_from_content, create_event_timestamp, + get_event_hash_from_content, ) timestamp = create_event_timestamp() diff --git a/syft_client/sync/version/__init__.py b/syft_client/sync/version/__init__.py index 7a45def3162..80cddae4cb5 100644 --- a/syft_client/sync/version/__init__.py +++ b/syft_client/sync/version/__init__.py @@ -5,20 +5,16 @@ Import it directly: from syft_client.sync.version.peer_manager import PeerManager """ -from syft_client.sync.version.version_info import VersionInfo from syft_client.sync.version.exceptions import ( VersionError, VersionMismatchError, VersionUnknownError, - ClientVersionMismatchError, - ProtocolVersionMismatchError, ) +from syft_client.sync.version.version_info import VersionInfo __all__ = [ - "VersionInfo", "VersionError", + "VersionInfo", "VersionMismatchError", "VersionUnknownError", - "ClientVersionMismatchError", - "ProtocolVersionMismatchError", ] diff --git a/syft_client/sync/version/exceptions.py b/syft_client/sync/version/exceptions.py index a4f7cecc932..f09ce30834a 100644 --- a/syft_client/sync/version/exceptions.py +++ b/syft_client/sync/version/exceptions.py @@ -11,8 +11,6 @@ class VersionError(Exception): """Base exception for version-related errors.""" - pass - class VersionMismatchError(VersionError): """Raised when versions are incompatible between peers.""" @@ -63,35 +61,3 @@ def __init__(self, peer_email: str, operation: Optional[str] = None): ) super().__init__(message) - - -class ClientVersionMismatchError(VersionMismatchError): - """Raised specifically for client version mismatches.""" - - def __init__( - self, - peer_email: str, - local_version: "VersionInfo", - peer_version: "VersionInfo", - ): - reason = ( - f"Client version mismatch: local={local_version.syft_client_version}, " - f"peer={peer_version.syft_client_version}" - ) - super().__init__(peer_email, local_version, peer_version, reason) - - -class ProtocolVersionMismatchError(VersionMismatchError): - """Raised specifically for protocol version mismatches.""" - - def __init__( - self, - peer_email: str, - local_version: "VersionInfo", - peer_version: "VersionInfo", - ): - reason = ( - f"Protocol version mismatch: local={local_version.protocol_version}, " - f"peer={peer_version.protocol_version}" - ) - super().__init__(peer_email, local_version, peer_version, reason) diff --git a/syft_client/sync/version/peer_manager.py b/syft_client/sync/version/peer_manager.py index fa635cb3e96..f07ffcef50c 100644 --- a/syft_client/sync/version/peer_manager.py +++ b/syft_client/sync/version/peer_manager.py @@ -99,8 +99,9 @@ class PeerManagerConfig(BaseModel): syftbox_folder: Path email: str = "" connection_configs: List[ConnectionConfig] = [] + # Applies to a peer of unknown version only. A client version difference does + # not skip a peer, so this flag has no effect on one. force_ignore_peer_version: bool = False - force_ignore_protocol_version: bool = True suppress_version_warnings: bool = False n_threads: int = 10 has_do_role: bool = False @@ -140,7 +141,6 @@ class PeerManager(BaseModel): connection_router: ConnectionRouter peer_store: PeerStore force_ignore_peer_version: bool = False - force_ignore_protocol_version: bool = True suppress_version_warnings: bool = False n_threads: int = 10 has_do_role: bool = False @@ -211,7 +211,6 @@ def from_config(cls, config: PeerManagerConfig, email: str = "") -> "PeerManager connection_router=connection_router, peer_store=peer_store, force_ignore_peer_version=config.force_ignore_peer_version, - force_ignore_protocol_version=config.force_ignore_protocol_version, suppress_version_warnings=config.suppress_version_warnings, n_threads=config.n_threads, has_do_role=config.has_do_role, @@ -370,11 +369,16 @@ def get_peer_compatibility_status( """Build a PeerCompatibilityResult describing whether the caller should skip / raise / warn for this peer. - SAME → no skip, no warning. PATCH_DIFF → no skip, "patch differs" - warning. INCOMPATIBLE / UNKNOWN → skip unless effective - `force_ignore_peer_version or ignore_peer_version` (then proceed with - a "proceeding to {action}" warning). UNKNOWN's skip message includes - a "call client.sync()" hint. + SAME → no skip, no log. + + PATCH_DIFF → no skip and a "patch differs" log, or a skip when + `skip_peer_on_patch_version_diff` is set. + + INCOMPATIBLE → no skip; the client version difference is logged, and + each protocol decides separately through its floor. + + UNKNOWN → skip, unless effective `force_ignore_peer_version or + ignore_peer_version`; the message includes a "call client.sync()" hint. """ own_version = self.get_own_version() peer_version = self.get_peer_version(peer_email) @@ -422,14 +426,26 @@ def get_peer_compatibility_status( **common, ) - # UNKNOWN or INCOMPATIBLE - if status == CompatibilityStatus.UNKNOWN: - detail = ( - "version information not available " - "(if you are unsure if it is up to date, call client.sync())" + if status == CompatibilityStatus.INCOMPATIBLE: + # A different client version does not refuse the peer. What each side + # can exchange is decided per protocol by the floor published in + # VersionInfo (MigrationRegistry.negotiate_protocol_version), not by + # comparing package versions. + return PeerCompatibilityResult( + should_skip=False, + explanation_not_skip=( + f"Peer {peer_email}: " + f"{own_version.get_incompatibility_reason(peer_version)}." + ), + **common, ) - else: - detail = own_version.get_incompatibility_reason(peer_version) + + # UNKNOWN: the capabilities of the peer are not known, so there is no + # floor to check. Skipping stays the safe answer. + detail = ( + "version information not available " + "(if you are unsure if it is up to date, call client.sync())" + ) effective_ignore = self.force_ignore_peer_version or ignore_peer_version if effective_ignore: @@ -485,8 +501,10 @@ def warn_if_all_peers_incompatible(self, peer_emails: List[str]) -> None: ) if not any_compatible: warnings.warn( - f"All connected peers ({len(peer_emails)}) have incompatible versions. " - "You may not be able to submit jobs or load datasets until versions match." + f"All connected peers ({len(peer_emails)}) run a different client " + "version, or their version is unknown. A peer with an unknown " + "version cannot receive jobs or datasets; call client.sync() to " + "read the version of each peer." ) def shutdown(self) -> None: diff --git a/syft_client/sync/version/version_info.py b/syft_client/sync/version/version_info.py index 28f1453c67d..73c6bf27862 100644 --- a/syft_client/sync/version/version_info.py +++ b/syft_client/sync/version/version_info.py @@ -157,6 +157,7 @@ def _slim_schema_of(registry) -> ProtocolSchema: return ProtocolSchema( protocol_name=registry.protocol_name, version=registry.protocol_version, + min_supported_version=registry.min_supported_protocol_version, supported_versions={ canonical_name: sorted(versions) for canonical_name, versions in registry.objects.items() diff --git a/tests/migrations/p2p/test_job_protocol_skew_delivery.py b/tests/migrations/p2p/test_job_protocol_skew_delivery.py new file mode 100644 index 00000000000..3d0b63ccb54 --- /dev/null +++ b/tests/migrations/p2p/test_job_protocol_skew_delivery.py @@ -0,0 +1,91 @@ +"""A job written for a protocol-0 peer reaches that peer and reads back. + +The other tests in this folder stop at the negotiated version. They assert which +protocol the two sides agree on, not that a job written at that protocol arrives +and reads. That seam is where the dataset transport broke: negotiation chose a +layout the delivery path could not carry. + +This test drives the whole path: the peer advertises job protocol 0, the sender +negotiates down, writes the flat layout, syncs, and the receiver finds and reads +the job through its own scan. +""" + +from pathlib import Path + +import pytest +from syft_client.sync.syftbox_manager import SyftboxManager +from syft_migration import ProtocolSchema + +from tests.unit.utils import create_test_project_folder + + +def _job_schema(protocol_version: str) -> ProtocolSchema: + # The slim form a peer advertises in its VersionInfo. + return ProtocolSchema( + protocol_name="syft-job", + version=protocol_version, + supported_versions={"JobState": ["1"], "JobSubmissionMetadata": ["1"]}, + ) + + +@pytest.fixture +def pair(): + return SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, + sync_automatically=False, + ) + + +def _submit(ds_manager, do_manager, job_name: str) -> Path: + project_dir = create_test_project_folder(with_pyproject=False) + ds_manager.submit_python_job( + user=do_manager.email, + code_path=str(project_dir), + job_name=job_name, + entrypoint="main.py", + ) + do_manager.sync() + return project_dir + + +def test_a_job_for_a_protocol0_peer_uses_the_flat_layout(pair): + ds_manager, do_manager = pair + # The DO advertises job protocol 0, as a client of 0.1.38 or earlier does. + ds_manager.peer_manager.live_peer_schemas("syft-job")[do_manager.email] = ( + _job_schema("0") + ) + + ref = ds_manager.job_client.manager.new_submission_ref(do_manager.email, "skew.job") + assert ref.protocol_version == "0" + assert "/v0/" not in str(ref) and "/v1/" not in str(ref), ( + "protocol 0 is the flat layout, so the path carries no v segment" + ) + + +def test_a_job_for_a_protocol0_peer_arrives_and_reads(pair): + ds_manager, do_manager = pair + ds_manager.peer_manager.live_peer_schemas("syft-job")[do_manager.email] = ( + _job_schema("0") + ) + + _submit(ds_manager, do_manager, "skew.job") + + # The receiver scans every layout it knows, so it finds the flat one. + assert [job.name for job in do_manager.jobs] == ["skew.job"] + found = do_manager.job_client.manager.find_submission_ref( + do_manager.email, "skew.job" + ) + assert found.protocol_version == "0" + + +def test_a_job_for_a_current_peer_still_uses_the_versioned_layout(pair): + # The control: without a protocol-0 peer the sender keeps the current layout, + # so the test above measures negotiation and not a broken default. + ds_manager, do_manager = pair + _submit(ds_manager, do_manager, "current.job") + + found = do_manager.job_client.manager.find_submission_ref( + do_manager.email, "current.job" + ) + assert found.protocol_version != "0" + assert [job.name for job in do_manager.jobs] == ["current.job"] diff --git a/tests/migrations/p2p/test_unknown_peer_forced_path.py b/tests/migrations/p2p/test_unknown_peer_forced_path.py new file mode 100644 index 00000000000..f680d87ce70 --- /dev/null +++ b/tests/migrations/p2p/test_unknown_peer_forced_path.py @@ -0,0 +1,75 @@ +"""A forced submission reports the protocol version that it assumes. + +A peer of unknown version is refused before this point. A caller that passes +``raise_on_unknown=False`` skips that refusal. The storage then assumes the +current protocol. + +If the peer speaks an earlier protocol, it does not scan this layout. The job or +the dataset never arrives, so the storage writes a warning. +""" + +import logging +from pathlib import Path + +from syft_datasets.config import SyftBoxConfig +from syft_datasets.dataset_storage import DatasetStorage +from syft_datasets.migrations.registry import DATASET_PROTOCOL_VERSION +from syft_job import SyftJobConfig +from syft_job.job_storage import JobStorage +from syft_job.migrations.registry import JOB_PROTOCOL_VERSION + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" + + +def _job_storage(tmp_path: Path) -> JobStorage: + config = SyftJobConfig( + syftbox_folder=tmp_path / "SyftBox", current_user_email=DS_EMAIL + ) + (tmp_path / "SyftBox" / DS_EMAIL).mkdir(parents=True, exist_ok=True) + return JobStorage(config=config, peer_schemas={}) + + +def _dataset_storage(tmp_path: Path) -> DatasetStorage: + config = SyftBoxConfig(syftbox_folder=tmp_path / "SyftBox", email=DO_EMAIL) + (tmp_path / "SyftBox" / DO_EMAIL).mkdir(parents=True, exist_ok=True) + return DatasetStorage(config=config, peer_schemas={}) + + +def test_a_forced_job_reports_the_assumed_protocol(tmp_path, caplog): + storage = _job_storage(tmp_path) + with caplog.at_level(logging.WARNING): + version = storage.negotiated_protocol_version_for_peer( + DO_EMAIL, raise_on_unknown=False + ) + assert version == JOB_PROTOCOL_VERSION + messages = " ".join(r.getMessage() for r in caplog.records) + assert DO_EMAIL in messages + assert "earlier protocol" in messages + + +def test_a_forced_dataset_reports_the_assumed_protocol(tmp_path, caplog): + storage = _dataset_storage(tmp_path) + with caplog.at_level(logging.WARNING): + version = storage.negotiated_protocol_version_for_peer( + DS_EMAIL, raise_on_unknown=False + ) + assert version == DATASET_PROTOCOL_VERSION + messages = " ".join(r.getMessage() for r in caplog.records) + assert DS_EMAIL in messages + assert "earlier protocol" in messages + + +def test_a_known_peer_reports_nothing(tmp_path, caplog): + # The report belongs to the forced path only. A known peer is negotiated. + from syft_migration import ProtocolSchema + + storage = _job_storage(tmp_path) + storage.peer_schemas[DO_EMAIL] = ProtocolSchema( + protocol_name="syft-job", + version=JOB_PROTOCOL_VERSION, + supported_versions={"JobState": ["1"]}, + ) + with caplog.at_level(logging.WARNING): + storage.negotiated_protocol_version_for_peer(DO_EMAIL) + assert caplog.records == [] diff --git a/tests/migrations/unit/test_history_artifacts.py b/tests/migrations/unit/test_history_artifacts.py index 6d9adcfeadb..98f507752f2 100644 --- a/tests/migrations/unit/test_history_artifacts.py +++ b/tests/migrations/unit/test_history_artifacts.py @@ -63,6 +63,16 @@ def test_protocol_not_changed_without_bump(): assert not client_registry.protocol_changed_without_bump() +def test_protocol_bump_not_missing(): + # Stays live between a protocol bump and the release that freezes it, which is + # exactly where test_protocol_not_changed_without_bump goes quiet. + assert not client_registry.protocol_bump_missing(), ( + "The client protocol changed since the newest released protocol without a " + "bump. Bump SYFT_CLIENT_PROTOCOL_VERSION in " + "syft_client/migrations/registry.py, or revert the model change." + ) + + def test_bump_guard_trips_on_protocol_change(): # A registry claiming the same protocol version as a released schema but # supporting different object versions must trip the guard. diff --git a/tests/migrations/unit/test_version_info_fields.py b/tests/migrations/unit/test_version_info_fields.py new file mode 100644 index 00000000000..5b9e5a2a1d1 --- /dev/null +++ b/tests/migrations/unit/test_version_info_fields.py @@ -0,0 +1,74 @@ +"""VersionInfo may only grow, because it is the bootstrap channel. + +A peer reads SYFT_version.json before it knows anything else, so every supported +client must parse every newer file. Two rules follow, and neither is enforced by +the migration system: + +- A field of an older version must not disappear or change name. An older reader + requires it, and pydantic raises when it is absent. +- A field that a newer version adds must have a default. A newer reader must + still parse a file that an older client wrote without that field. + +Adding a field is safe on its own: pydantic ignores a field it does not know. +""" + +import syft_client # noqa: F401 -- imports models and registers history +from syft_client.sync.version.version_info import VersionInfoV1, VersionInfoV2 + +# Frozen on purpose. A change here means a change to the bootstrap file, so read +# the two rules above before editing this set. +V1_FIELDS = { + "canonical_name", + "version", + "syft_client_version", + "min_supported_syft_client_version", + "protocol_version", + "min_supported_protocol_version", + "syft_client_install_source", + "updated_at", + "attestation_token", +} + +V2_ADDS = {"protocol_schemas"} + + +def test_v1_fields_are_frozen(): + assert set(VersionInfoV1.model_fields) == V1_FIELDS, ( + "VersionInfoV1 changed. A client that speaks protocol 0 reads this " + "object, so a removed or renamed field stops that client from parsing " + "the version file of this one." + ) + + +def test_v2_keeps_every_v1_field(): + missing = V1_FIELDS - set(VersionInfoV2.model_fields) + assert not missing, ( + f"VersionInfoV2 dropped {sorted(missing)}. A reader of V1 requires these " + "fields, so V2 must keep them." + ) + + +def test_v2_adds_only_the_expected_fields(): + assert set(VersionInfoV2.model_fields) - V1_FIELDS == V2_ADDS + + +def test_fields_added_after_v1_have_a_default(): + # A file written by an older client carries none of these, so a reader of the + # newer version must supply a value. + for name in set(VersionInfoV2.model_fields) - V1_FIELDS: + assert not VersionInfoV2.model_fields[name].is_required(), ( + f"VersionInfoV2.{name} is required. A version file written before " + "this field existed would then fail to parse." + ) + + +def test_a_file_without_the_v2_fields_still_parses(): + written_by_an_older_client = VersionInfoV1( + syft_client_version="0.1.117", + min_supported_syft_client_version="0.1.93", + protocol_version="1.0.0", + min_supported_protocol_version="1.0.0", + ).model_dump(exclude={"canonical_name", "version"}) + + loaded = VersionInfoV2.model_validate(written_by_an_older_client) + assert loaded.protocol_schemas == {} diff --git a/tests/unit/test_bump_version.py b/tests/unit/test_bump_version.py new file mode 100644 index 00000000000..2e6508f2864 --- /dev/null +++ b/tests/unit/test_bump_version.py @@ -0,0 +1,88 @@ +"""Check the version that bump_version.py writes into the pin of a dependent.""" + +import importlib.util +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "scripts" / "bump_version.py" + +TARGET = """\ +[project] +name = "syft-thing" +version = "0.1.9" +dependencies = [] +""" + +DEPENDENT = """\ +[project] +name = "syft-other" +version = "0.2.0" +dependencies = [ + "syft-thing==0.1.9", +] + +[tool.uv.sources] +"syft-thing" = { workspace = true } +""" + + +@pytest.fixture +def fake_repo(tmp_path): + (tmp_path / "packages" / "syft-thing").mkdir(parents=True) + (tmp_path / "packages" / "syft-other").mkdir(parents=True) + (tmp_path / "packages" / "syft-thing" / "pyproject.toml").write_text(TARGET) + (tmp_path / "packages" / "syft-other" / "pyproject.toml").write_text(DEPENDENT) + return tmp_path + + +def _run(fake_repo, *args): + spec = importlib.util.spec_from_file_location("bump_version_under_test", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + module.REPO_ROOT = fake_repo + argv = [str(SCRIPT), "syft-thing", "patch", *args] + old = sys.argv + sys.argv = argv + try: + module.main() + finally: + sys.argv = old + + +def _versions(fake_repo): + target = (fake_repo / "packages" / "syft-thing" / "pyproject.toml").read_text() + dependent = (fake_repo / "packages" / "syft-other" / "pyproject.toml").read_text() + source = next( + line for line in target.splitlines() if line.startswith("version") + ).split('"')[1] + pin = next(line for line in dependent.splitlines() if "syft-thing==" in line) + return source, pin.split("==")[1].split('"')[0] + + +def test_default_pins_dependents_to_the_bumped_version(fake_repo): + _run(fake_repo) + source, pin = _versions(fake_repo) + assert source == "0.1.10" + assert pin == "0.1.10" + + +def test_published_pins_dependents_to_the_version_just_released(fake_repo): + # A release publishes the version on the branch, then bumps the version. The + # monorepo releases a dependent later in the same run. The pin must therefore + # name a version that PyPI already has. + _run(fake_repo, "--dependents", "published") + source, pin = _versions(fake_repo) + assert source == "0.1.10" + assert pin == "0.1.9" + + +def test_dependent_pin_is_a_published_version_for_every_release_order(fake_repo): + # This test covers the monorepo order. syft-perms releases before syft-job. If + # the script pins a dependent to the new version, syft-job publishes a + # dependency that PyPI does not have. + _run(fake_repo, "--dependents", "published") + _, pin = _versions(fake_repo) + assert pin == "0.1.9", "a dependent must pin the version that the release published" diff --git a/tests/unit/test_checkpoint_version.py b/tests/unit/test_checkpoint_version.py new file mode 100644 index 00000000000..38850d3c8d0 --- /dev/null +++ b/tests/unit/test_checkpoint_version.py @@ -0,0 +1,77 @@ +"""A checkpoint or rolling state from a later client is refused, not restored. + +Both models carry a `version` field that nothing read. A later client can change +what a field means while the object still parses, because pydantic accepts a +document that holds every field it knows. The restore would then be wrong and +silent. + +Refusing is cheap here. Every load site already falls back to a download of all +events when a checkpoint fails to load, so an unusable checkpoint costs one slow +cold start and nothing else. +""" + +import pytest +from syft_client.sync.checkpoints.checkpoint import ( + CHECKPOINT_VERSION, + Checkpoint, + IncrementalCheckpoint, +) +from syft_client.sync.checkpoints.rolling_state import ( + ROLLING_STATE_VERSION, + RollingState, +) + +EMAIL = "alice@example.com" + + +def _checkpoint(**kwargs) -> Checkpoint: + return Checkpoint(email=EMAIL, **kwargs) + + +def _incremental(**kwargs) -> IncrementalCheckpoint: + return IncrementalCheckpoint(email=EMAIL, sequence_number=1, **kwargs) + + +def _rolling(**kwargs) -> RollingState: + return RollingState(email=EMAIL, base_checkpoint_timestamp=1.0, **kwargs) + + +def test_a_checkpoint_round_trips(): + loaded = Checkpoint.from_compressed_data(_checkpoint().as_compressed_data()) + assert loaded.version == CHECKPOINT_VERSION + + +def test_an_incremental_checkpoint_round_trips(): + loaded = IncrementalCheckpoint.from_compressed_data( + _incremental().as_compressed_data() + ) + assert loaded.version == CHECKPOINT_VERSION + + +def test_a_rolling_state_round_trips(): + loaded = RollingState.from_compressed_data(_rolling().as_compressed_data()) + assert loaded.version == ROLLING_STATE_VERSION + + +def test_a_later_checkpoint_is_refused(): + data = _checkpoint(version=CHECKPOINT_VERSION + 1).as_compressed_data() + with pytest.raises(ValueError, match=str(CHECKPOINT_VERSION + 1)): + Checkpoint.from_compressed_data(data) + + +def test_a_later_incremental_checkpoint_is_refused(): + data = _incremental(version=CHECKPOINT_VERSION + 1).as_compressed_data() + with pytest.raises(ValueError, match=str(CHECKPOINT_VERSION + 1)): + IncrementalCheckpoint.from_compressed_data(data) + + +def test_a_later_rolling_state_is_refused(): + data = _rolling(version=ROLLING_STATE_VERSION + 1).as_compressed_data() + with pytest.raises(ValueError, match=str(ROLLING_STATE_VERSION + 1)): + RollingState.from_compressed_data(data) + + +def test_an_earlier_version_still_loads(): + # Version 0 predates the field. Those objects are the shape this client reads. + data = _checkpoint(version=0).as_compressed_data() + assert Checkpoint.from_compressed_data(data).version == 0 diff --git a/tests/unit/test_crypto_keys_version.py b/tests/unit/test_crypto_keys_version.py new file mode 100644 index 00000000000..bf32ff607d8 --- /dev/null +++ b/tests/unit/test_crypto_keys_version.py @@ -0,0 +1,51 @@ +"""The crypto key file carries a version, and an unknown one stops the load. + +A user cannot rebuild a private key, so delete-and-rebuild is not a recovery +here. If a newer client wrote the file, this client must refuse it rather than +read it wrong and lose the keys. +""" + +import json + +import pytest +from syft_client.sync.peers.peer_store import CRYPTO_KEYS_VERSION, PeerStore + + +def _saved(tmp_path): + store = PeerStore(email="alice@example.com", use_encryption=True) + store.generate_keys() + path = tmp_path / "crypto_keys.json" + store.save_keys(path) + return path + + +def test_a_saved_file_carries_the_version(tmp_path): + data = json.loads(_saved(tmp_path).read_text()) + assert data["version"] == CRYPTO_KEYS_VERSION + + +def test_a_saved_file_loads_back(tmp_path): + path = _saved(tmp_path) + loaded = PeerStore.load_keys(path) + assert loaded.email == "alice@example.com" + + +def test_a_file_without_a_version_still_loads(tmp_path): + # Written before the version field existed. Those keys must keep working. + path = _saved(tmp_path) + data = json.loads(path.read_text()) + del data["version"] + path.write_text(json.dumps(data)) + + loaded = PeerStore.load_keys(path) + assert loaded.email == "alice@example.com" + + +def test_a_file_from_a_newer_client_is_refused(tmp_path): + path = _saved(tmp_path) + data = json.loads(path.read_text()) + data["version"] = CRYPTO_KEYS_VERSION + 1 + path.write_text(json.dumps(data)) + + with pytest.raises(ValueError, match=str(CRYPTO_KEYS_VERSION + 1)): + PeerStore.load_keys(path) diff --git a/tests/unit/test_dataset_collection_listing.py b/tests/unit/test_dataset_collection_listing.py new file mode 100644 index 00000000000..606f0850a95 --- /dev/null +++ b/tests/unit/test_dataset_collection_listing.py @@ -0,0 +1,68 @@ +"""owner_list_all_dataset_collections_with_permissions skips only bad names. + +The Drive query matches a name prefix, so another tool can return a folder that +this client cannot parse. The listing skips that folder. Every other failure is a +defect, so the listing must raise it. +""" + +from unittest.mock import Mock + +import pytest + +from syft_client.sync.connections.drive.gdrive_transport import ( + DATASET_COLLECTION_PREFIX, + GDriveConnection, +) + +VALID = f"{DATASET_COLLECTION_PREFIX}_mytag_abc123" +UNPARSEABLE = DATASET_COLLECTION_PREFIX + + +def _conn(files): + conn = GDriveConnection(email="alice@example.com", verbose=False) + conn.drive_service = Mock() + conn._syftbox_folder_id = "syftbox-id" + conn.drive_service.files().list().execute.return_value = {"files": files} + return conn + + +def test_a_valid_collection_is_returned(): + conn = _conn([{"id": "f1", "name": VALID, "appProperties": {}}]) + got = conn.owner_list_all_dataset_collections_with_permissions() + assert [(c.folder_id, c.tag, c.content_hash) for c in got] == [ + ("f1", "mytag", "abc123") + ] + assert got[0].has_any_permission is False + + +def test_the_any_permission_flag_comes_from_app_properties(): + conn = _conn( + [ + { + "id": "f1", + "name": VALID, + "appProperties": {"syft_shared_with_any": "true"}, + } + ] + ) + got = conn.owner_list_all_dataset_collections_with_permissions() + assert got[0].has_any_permission is True + + +def test_a_name_the_client_cannot_parse_is_skipped(): + conn = _conn( + [ + {"id": "bad", "name": UNPARSEABLE, "appProperties": {}}, + {"id": "f1", "name": VALID, "appProperties": {}}, + ] + ) + got = conn.owner_list_all_dataset_collections_with_permissions() + assert [c.folder_id for c in got] == ["f1"] + + +def test_a_missing_name_field_raises(): + # A blanket except turned this defect into a collection that disappears + # without a message. + conn = _conn([{"id": "f1", "appProperties": {}}]) + with pytest.raises(KeyError): + conn.owner_list_all_dataset_collections_with_permissions() diff --git a/tests/unit/test_p2p_folder_lookup.py b/tests/unit/test_p2p_folder_lookup.py new file mode 100644 index 00000000000..0e8d79c7659 --- /dev/null +++ b/tests/unit/test_p2p_folder_lookup.py @@ -0,0 +1,90 @@ +"""P2P folder lookup accepts any client version in the folder name. + +A P2P folder name is a rendezvous string that both peers compute from their own +client version, so neither side may rename it (see the adopt path for private +folders). Lookup therefore has to tolerate the version instead. + +Reuse matters in both directions. A folder this client owns must be reused after +an upgrade, because a peer that still filters by name would not find a new one. +A folder the peer owns must be found whatever version the peer wrote into it. +""" + +from unittest.mock import Mock + +from syft_client.sync.connections.drive.gdrive_transport import GDriveConnection + +ME = "alice@example.com" +PEER = "bob@example.com" + + +def _name(version: str, datasite: str, folder_type: str, peer: str) -> str: + return f"syft_datasite#{version}#{datasite}#{folder_type}#{peer}" + + +def _conn(found): + conn = GDriveConnection(email=ME, verbose=False) + conn.drive_service = Mock() + conn._find_folders = Mock(return_value=found) + return conn + + +def _lookup(conn): + return conn._find_p2p_folder_id( + datasite_email=PEER, folder_type="inbox", peer_email=ME, owner_email=ME + ) + + +def test_a_folder_of_another_minor_version_is_found(): + # The old filter dropped this folder, so the client created a second one and + # the peer kept writing into the first. 0.2.0 differs in the minor from the + # current client version, which is what the filter used to reject. + old = _name("0.2.0", PEER, "inbox", ME) + assert _lookup(_conn([("old", old)])) == "old" + + +def test_a_folder_of_an_older_major_version_is_found(): + old = _name("0.0.9", PEER, "inbox", ME) + assert _lookup(_conn([("old", old)])) == "old" + + +def test_the_highest_version_wins_when_several_exist(): + folders = [ + ("v1", _name("0.1.117", PEER, "inbox", ME)), + ("v2", _name("0.2.0", PEER, "inbox", ME)), + ("v0", _name("0.0.9", PEER, "inbox", ME)), + ] + assert _lookup(_conn(folders)) == "v2" + + +def test_versions_order_by_number_not_by_string(): + folders = [ + ("nine", _name("0.1.9", PEER, "inbox", ME)), + ("ten", _name("0.1.10", PEER, "inbox", ME)), + ] + assert _lookup(_conn(folders)) == "ten" + + +def test_several_folders_no_longer_raise(): + folders = [ + ("a", _name("0.1.117", PEER, "inbox", ME)), + ("b", _name("0.1.118", PEER, "inbox", ME)), + ] + assert _lookup(_conn(folders)) is not None + + +def test_a_folder_of_another_peer_is_ignored(): + other = _name("0.1.117", PEER, "inbox", "carol@example.com") + assert _lookup(_conn([("other", other)])) is None + + +def test_a_folder_of_another_type_is_ignored(): + outbox = _name("0.1.117", PEER, "outbox", ME) + assert _lookup(_conn([("outbox", outbox)])) is None + + +def test_no_folder_returns_none(): + assert _lookup(_conn([])) is None + + +def test_a_name_that_does_not_parse_is_ignored(): + assert _lookup(_conn([("junk", "not_a_p2p_folder")])) is None diff --git a/tests/unit/test_peers_json_version.py b/tests/unit/test_peers_json_version.py new file mode 100644 index 00000000000..f8b09f9dc6b --- /dev/null +++ b/tests/unit/test_peers_json_version.py @@ -0,0 +1,101 @@ +"""SYFT_peers.json carries a version, and an unreadable peer state is logged. + +The file is a flat map of peer email to entry, so a version cannot go at the top +level: every existing client reads a top-level key as an email. The version lives +under a reserved key instead. An older client parses the state of that entry, +fails, and skips it, so the reserved key is invisible to a client that predates +it. + +The record itself is safe either way. The only writer is `_update_peer_state`, +which changes one entry of the raw map and writes the rest back, so a peer this +client cannot read is not erased for the other side. +""" + +import logging +from unittest.mock import Mock, patch + +from syft_client.sync.connections.drive.gdrive_transport import ( + PEERS_META_KEY, + SYFT_PEERS_VERSION, + GDriveConnection, +) +from syft_client.sync.peers.peer import PeerState + +PEER = "bob@example.com" + + +def _conn(peers_data): + conn = GDriveConnection(email="alice@example.com", verbose=False) + conn.drive_service = Mock() + conn._peers_json_cache = dict(peers_data) + return conn + + +def _router(conn): + router = Mock() + router.connection_for_send_message = Mock(return_value=conn) + from syft_client.sync.connections.connection_router import ConnectionRouter + + return ConnectionRouter.get_all_peers_from_json.__get__(router, ConnectionRouter) + + +def test_a_write_stamps_the_reserved_entry(): + conn = _conn({PEER: {"state": "accepted"}}) + with ( + patch.object(GDriveConnection, "_get_peers_file_id", return_value="file-id"), + patch.object( + GDriveConnection, "get_syftbox_folder_id", return_value="folder-id" + ), + patch.object( + GDriveConnection, "create_file_payload", return_value=(Mock(), None) + ), + ): + conn._write_peers_json({PEER: {"state": "accepted"}}) + + assert conn._peers_json_cache[PEERS_META_KEY] == {"version": SYFT_PEERS_VERSION} + assert conn._peers_json_cache[PEER] == {"state": "accepted"} + + +def test_the_reserved_entry_is_not_a_peer(): + conn = _conn( + { + PEERS_META_KEY: {"version": SYFT_PEERS_VERSION}, + PEER: {"state": "accepted"}, + } + ) + peers = _router(conn)() + assert [p.email for p in peers] == [PEER] + + +def test_a_known_state_loads(): + conn = _conn({PEER: {"state": "rejected"}}) + peers = _router(conn)() + assert peers[0].state == PeerState.REJECTED + + +def test_an_unknown_state_is_skipped_and_logged(caplog): + conn = _conn({PEER: {"state": "quarantined"}}) + with caplog.at_level(logging.WARNING, logger="syft_client"): + peers = _router(conn)() + assert peers == [] + assert any(PEER in r.getMessage() for r in caplog.records) + assert any("quarantined" in r.getMessage() for r in caplog.records) + + +def test_a_file_without_the_reserved_entry_still_loads(): + # Written before the reserved key existed. + conn = _conn({PEER: {"state": "accepted"}}) + peers = _router(conn)() + assert [p.email for p in peers] == [PEER] + + +def test_a_reserved_entry_from_a_newer_client_does_not_stop_the_read(caplog): + conn = _conn( + { + PEERS_META_KEY: {"version": SYFT_PEERS_VERSION + 1}, + PEER: {"state": "accepted"}, + } + ) + with caplog.at_level(logging.WARNING, logger="syft_client"): + peers = _router(conn)() + assert [p.email for p in peers] == [PEER] diff --git a/tests/unit/test_persisted_dict.py b/tests/unit/test_persisted_dict.py index b8a72a576f8..650d11d3ecd 100644 --- a/tests/unit/test_persisted_dict.py +++ b/tests/unit/test_persisted_dict.py @@ -36,7 +36,7 @@ def writer(d: PersistedDict, prefix: str): assert errors == [], f"Concurrent writes raised: {errors!r}" # Every key from both writers must be present in the final on-disk state. - final = json.loads(target.read_text()) + final = json.loads(target.read_text())["entries"] expected = {f"a-{i}": i for i in range(iterations)} | { f"b-{i}": i for i in range(iterations) } @@ -60,7 +60,7 @@ def test_set_with_write_false_does_not_persist(tmp_path: Path): with d.exclusive_lock(): d._write_to_file() - assert json.loads(target.read_text()) == {"k": "v"} + assert json.loads(target.read_text())["entries"] == {"k": "v"} def test_batch_write_with_exclusive_lock(tmp_path: Path): @@ -86,7 +86,7 @@ def batch_write(d: PersistedDict, prefix: str, n: int): t1.join() t2.join() - final = json.loads(target.read_text()) + final = json.loads(target.read_text())["entries"] expected = {f"a-{i}": i for i in range(50)} | {f"b-{i}": i for i in range(50)} assert final == expected @@ -108,4 +108,4 @@ def test_contains_and_delete_with_flags(tmp_path: Path): d._write_to_file() # After the batch, on-disk state reflects the in-memory delete. - assert json.loads(target.read_text()) == {} + assert json.loads(target.read_text())["entries"] == {} diff --git a/tests/unit/test_persisted_dict_version.py b/tests/unit/test_persisted_dict_version.py new file mode 100644 index 00000000000..44aba4f97e3 --- /dev/null +++ b/tests/unit/test_persisted_dict_version.py @@ -0,0 +1,65 @@ +"""A persisted cache carries a version, and an unknown one resets the cache. + +The client can rebuild every one of these caches from the events and the files, +so an unreadable cache costs a re-scan and nothing else. An unknown version +therefore starts empty instead of stopping the client. +""" + +import json + +from syft_client.sync.sync.caches.persisted_dict import ( + PERSISTED_DICT_VERSION, + PersistedDict, +) + + +def _path(tmp_path): + return tmp_path / "cache.json" + + +def test_a_saved_file_carries_the_version(tmp_path): + d = PersistedDict(path=_path(tmp_path)) + d["a"] = "1" + data = json.loads(_path(tmp_path).read_text()) + assert data["version"] == PERSISTED_DICT_VERSION + assert data["entries"] == {"a": "1"} + + +def test_a_saved_file_loads_back(tmp_path): + d = PersistedDict(path=_path(tmp_path)) + d["a"] = "1" + assert PersistedDict(path=_path(tmp_path)).get("a") == "1" + + +def test_a_file_without_a_version_still_loads(tmp_path): + # Written before the version field existed: a bare map of entries. Reading it + # saves the user a full re-scan on the first run after an upgrade. + _path(tmp_path).write_text(json.dumps({"a": "1", "b": "2"})) + d = PersistedDict(path=_path(tmp_path)) + assert d.get("a") == "1" + assert d.get("b") == "2" + + +def test_a_file_from_a_newer_client_starts_empty(tmp_path): + _path(tmp_path).write_text( + json.dumps({"version": PERSISTED_DICT_VERSION + 1, "entries": {"a": "1"}}) + ) + d = PersistedDict(path=_path(tmp_path)) + assert d.get("a") is None + assert len(d) == 0 + + +def test_an_unreadable_file_starts_empty(tmp_path): + _path(tmp_path).write_text("{not json") + assert len(PersistedDict(path=_path(tmp_path))) == 0 + + +def test_a_reset_cache_can_be_written_again(tmp_path): + _path(tmp_path).write_text( + json.dumps({"version": PERSISTED_DICT_VERSION + 1, "entries": {"a": "1"}}) + ) + d = PersistedDict(path=_path(tmp_path)) + d["b"] = "2" + data = json.loads(_path(tmp_path).read_text()) + assert data["version"] == PERSISTED_DICT_VERSION + assert data["entries"] == {"b": "2"} diff --git a/tests/unit/test_version_mismatch_flow.py b/tests/unit/test_version_mismatch_flow.py index 471142d0e71..75f2d9b8fa1 100644 --- a/tests/unit/test_version_mismatch_flow.py +++ b/tests/unit/test_version_mismatch_flow.py @@ -2,7 +2,6 @@ from unittest.mock import patch -from syft_client.sync.utils.syftbox_utils import delete_local_syftbox from syft_client.sync.connections.drive.gdrive_transport import ( GDRIVE_P2P_FOLDER_DATASITE_PREFIX, GOOGLE_FOLDER_MIME_TYPE, @@ -12,6 +11,7 @@ MockDriveService, ) from syft_client.sync.syftbox_manager import SyftboxManager, SyftboxManagerConfig +from syft_client.sync.utils.syftbox_utils import delete_local_syftbox from syft_client.version import SYFT_CLIENT_VERSION from tests.unit.utils import create_test_project_folder, create_tmp_dataset_files @@ -240,12 +240,23 @@ def test_version_mismatch_and_backup_flow(): do_manager.load_peers() do_manager.approve_peer_request(ds_manager.email) - # Now new versioned P2P folders should exist + # The P2P folders of the old version are reused, not replaced. Both + # peers compute this folder name from their own client version, so a + # peer that has not upgraded still looks for the old name. A second + # folder under NEW_VERSION would hide the first one from that peer. do_p2p_new = _find_versioned_p2p_folders(do_conn_new, ds_email, NEW_VERSION) - assert len(do_p2p_new) > 0 + assert len(do_p2p_new) == 0 + do_p2p_old = _find_versioned_p2p_folders( + do_conn_new, ds_email, SYFT_CLIENT_VERSION + ) + assert len(do_p2p_old) > 0 ds_p2p_new = _find_versioned_p2p_folders(ds_conn_new, do_email, NEW_VERSION) - assert len(ds_p2p_new) > 0 + assert len(ds_p2p_new) == 0 + ds_p2p_old = _find_versioned_p2p_folders( + ds_conn_new, do_email, SYFT_CLIENT_VERSION + ) + assert len(ds_p2p_old) > 0 # -- Step 14: Re-upload dataset -- mock_path2, private_path2, readme_path2 = create_tmp_dataset_files() diff --git a/tests/unit/test_version_negotiation.py b/tests/unit/test_version_negotiation.py index 36bb593f4ac..339ca4c7e4d 100644 --- a/tests/unit/test_version_negotiation.py +++ b/tests/unit/test_version_negotiation.py @@ -5,7 +5,6 @@ import pytest from syft_client.sync.syftbox_manager import SyftboxManager from syft_client.sync.version.exceptions import ( - VersionMismatchError, VersionUnknownError, ) from syft_client.sync.version.peer_manager import CompatAction @@ -455,33 +454,36 @@ def test_explicit_true_on_ds_is_preserved(self): class TestForceAllowIncompatiblePeers: """Tests for force_ignore_peer_version and per-call ignore_peer_version.""" - def test_incompatible_peer_skipped_by_default(self): + def test_incompatible_peer_is_included_with_a_log(self, caplog): + # A different client version no longer refuses a peer. The protocol floor + # in VersionInfo decides what the two sides may exchange. ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( check_versions=True, ) _set_peer_version(do_manager, ds_manager.email, build_client_version("99.0.0")) - do_manager.peer_manager.suppress_version_warnings = True - compatible = do_manager.peer_manager.get_compatible_peer_emails_for_syncing( - [ds_manager.email] + with caplog.at_level(logging.INFO, logger="syft_client"): + compatible = do_manager.peer_manager.get_compatible_peer_emails_for_syncing( + [ds_manager.email] + ) + assert ds_manager.email in compatible + assert any( + "client version mismatch" in r.getMessage().lower() for r in caplog.records ) - assert ds_manager.email not in compatible - def test_force_allow_includes_incompatible_peer(self, caplog): + def test_force_allow_is_redundant_for_an_incompatible_peer(self): + # The flag overrode a refusal that no longer happens. The peer is included + # either way. ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( check_versions=True, ) _set_peer_version(do_manager, ds_manager.email, build_client_version("99.0.0")) do_manager.peer_manager.force_ignore_peer_version = True - with caplog.at_level(logging.INFO, logger="syft_client"): - compatible = do_manager.peer_manager.get_compatible_peer_emails_for_syncing( - [ds_manager.email] - ) - assert ds_manager.email in compatible - assert any( - "proceeding anyway" in r.getMessage().lower() for r in caplog.records + compatible = do_manager.peer_manager.get_compatible_peer_emails_for_syncing( + [ds_manager.email] ) + assert ds_manager.email in compatible def test_per_call_ignore_peer_version_includes_peer(self): ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( @@ -494,19 +496,22 @@ def test_per_call_ignore_peer_version_includes_peer(self): ) assert ds_manager.email in compatible - def test_per_call_ignore_peer_version_in_submit(self): + def test_submit_no_longer_raises_for_an_incompatible_peer(self): + # A client version difference does not stop a submission. Only an unknown + # peer version does (see test_job_submission_blocked_without_version). ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( check_versions=True, ) _set_peer_version(ds_manager, do_manager.email, build_client_version("99.0.0")) - with pytest.raises(VersionMismatchError): - result = ds_manager.peer_manager.get_peer_compatibility_status( - do_manager.email, action=CompatAction.SUBMIT - ) - result.raise_on_skip(operation="submit job") + result = ds_manager.peer_manager.get_peer_compatibility_status( + do_manager.email, action=CompatAction.SUBMIT + ) + assert result.status == CompatibilityStatus.INCOMPATIBLE + assert not result.should_skip + result.raise_on_skip(operation="submit job") - # With per-call override, should not raise + # The per-call override is redundant now, and still does not raise. result = ds_manager.peer_manager.get_peer_compatibility_status( do_manager.email, action=CompatAction.SUBMIT, @@ -531,12 +536,31 @@ def test_force_allow_in_submit(self): class TestVersionMismatchBehavior: """Tests for version mismatch behavior during operations.""" - def test_sync_skips_incompatible_peers(self): + def test_sync_keeps_an_incompatible_peer(self): ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( check_versions=True, ) _set_peer_version(do_manager, ds_manager.email, build_client_version("0.0.1")) + do_manager.peer_manager.suppress_version_warnings = True + compatible_peers = ( + do_manager.peer_manager.get_compatible_peer_emails_for_syncing( + [ds_manager.email] + ) + ) + assert ds_manager.email in compatible_peers + + def test_sync_still_skips_a_peer_of_unknown_version(self): + # The boundary of the policy: a known difference is allowed, an unknown + # peer is not. Nothing can be negotiated without the version of the peer. + ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( + check_versions=True, + ) + peer = do_manager.peer_manager.get_cached_peer(ds_manager.email) + assert peer is not None + peer.version = None + do_manager.peer_manager._loaded_peer_versions[ds_manager.email] = None + do_manager.peer_manager.suppress_version_warnings = True compatible_peers = ( do_manager.peer_manager.get_compatible_peer_emails_for_syncing( diff --git a/tests/unit/test_versioned_folder_adopt.py b/tests/unit/test_versioned_folder_adopt.py new file mode 100644 index 00000000000..7dac544a0ed --- /dev/null +++ b/tests/unit/test_versioned_folder_adopt.py @@ -0,0 +1,134 @@ +"""A client adopts a private Drive folder from an earlier client version. + +A private folder name holds the client version. After a minor upgrade the name of +the current version does not exist yet. Without adoption the client creates a new +folder, and the datasite of the user stays on Drive out of reach. + +These tests cover the private folders only. The name of a P2P folder is a +rendezvous string that both peers compute, so a client must never rename one. +""" + +from unittest.mock import Mock + +import pytest + +from syft_client.sync.connections.drive.gdrive_transport import ( + GDriveConnection, + _partition_by_version, +) + +EMAIL = "alice@example.com" + + +def _conn(): + conn = GDriveConnection(email=EMAIL, verbose=False) + conn.drive_service = Mock() + return conn + + +def _renames(conn): + """Return the (fileId, new name) pairs the connection sent to Drive.""" + return [ + (kwargs["fileId"], kwargs["body"]["name"]) + for _, kwargs in conn.drive_service.files().update.call_args_list + if "body" in kwargs and "name" in kwargs.get("body", {}) + ] + + +# ---------- _partition_by_version ------------------------------------------- + + +def test_partition_splits_compatible_older_and_newer(): + folders = [ + ("old", f"0.1.9#{EMAIL}"), + ("same", f"0.2.5#{EMAIL}"), + ("new", f"0.3.0#{EMAIL}"), + ] + compatible, older, newer = _partition_by_version(folders, current_version="0.2.7") + assert compatible == [("same", f"0.2.5#{EMAIL}")] + assert older == [("old", f"0.1.9#{EMAIL}")] + assert newer == [("new", f"0.3.0#{EMAIL}")] + + +def test_partition_sorts_by_number_not_by_string(): + folders = [("a", f"0.1.9#{EMAIL}"), ("b", f"0.1.10#{EMAIL}")] + _, older, _ = _partition_by_version(folders, current_version="0.2.0") + assert [fid for fid, _ in older] == ["a", "b"] + + +def test_partition_drops_names_without_a_version(): + folders = [("a", f"0.1.9#{EMAIL}"), ("b", "no_version_here")] + _, older, _ = _partition_by_version(folders, current_version="0.2.0") + assert older == [("a", f"0.1.9#{EMAIL}")] + + +def test_partition_returns_empty_for_a_bad_current_version(): + folders = [("a", f"0.1.9#{EMAIL}")] + assert _partition_by_version(folders, current_version="garbage") == ([], [], []) + + +# ---------- adoption -------------------------------------------------------- + + +def test_a_compatible_folder_wins_and_nothing_is_renamed(): + conn = _conn() + folders = [("same", f"0.2.5#{EMAIL}"), ("old", f"0.1.9#{EMAIL}")] + got = conn._find_or_adopt_versioned_folder( + folders, current_name=f"0.2.7#{EMAIL}", current_version="0.2.7" + ) + assert got == "same" + assert _renames(conn) == [] + + +def test_an_older_folder_is_adopted_by_rename(): + conn = _conn() + folders = [("old", f"0.1.9#{EMAIL}")] + got = conn._find_or_adopt_versioned_folder( + folders, current_name=f"0.2.7#{EMAIL}", current_version="0.2.7" + ) + assert got == "old", "the client must keep the folder that holds the data" + assert _renames(conn) == [("old", f"0.2.7#{EMAIL}")] + + +def test_the_highest_older_folder_is_adopted(): + conn = _conn() + folders = [ + ("v1", f"0.1.9#{EMAIL}"), + ("v2", f"0.1.20#{EMAIL}"), + ("v0", f"0.0.4#{EMAIL}"), + ] + got = conn._find_or_adopt_versioned_folder( + folders, current_name=f"0.2.7#{EMAIL}", current_version="0.2.7" + ) + assert got == "v2" + assert _renames(conn) == [("v2", f"0.2.7#{EMAIL}")] + + +def test_a_newer_folder_stops_the_client(): + # A new folder here would hide data that this client cannot read. Report the + # version to install instead. + conn = _conn() + folders = [("new", f"0.3.0#{EMAIL}")] + with pytest.raises(RuntimeError, match="0.3.0"): + conn._find_or_adopt_versioned_folder( + folders, current_name=f"0.2.7#{EMAIL}", current_version="0.2.7" + ) + assert _renames(conn) == [] + + +def test_no_folder_returns_none_so_the_caller_creates_one(): + conn = _conn() + got = conn._find_or_adopt_versioned_folder( + [], current_name=f"0.2.7#{EMAIL}", current_version="0.2.7" + ) + assert got is None + assert _renames(conn) == [] + + +def test_two_compatible_folders_still_raise(): + conn = _conn() + folders = [("a", f"0.2.1#{EMAIL}"), ("b", f"0.2.2#{EMAIL}")] + with pytest.raises(RuntimeError): + conn._find_or_adopt_versioned_folder( + folders, current_name=f"0.2.7#{EMAIL}", current_version="0.2.7" + ) diff --git a/tests/unit/test_versioned_folder_lookup.py b/tests/unit/test_versioned_folder_lookup.py index f9dce12b8d2..0d3176cd7f2 100644 --- a/tests/unit/test_versioned_folder_lookup.py +++ b/tests/unit/test_versioned_folder_lookup.py @@ -2,15 +2,16 @@ These are pure functions -- no Drive mocks needed. They cover the path that replaced the four format-specific parsers from the original PR. + +Ordering and selection now live in _partition_by_version (adopt, private +folders) and _sorted_by_version (P2P lookup), each tested separately. """ from syft_client.sync.connections.drive.gdrive_transport import ( _extract_version_from_name, - _filter_patch_compatible, _looks_like_version, ) - # ---------- _looks_like_version --------------------------------------------- @@ -61,63 +62,3 @@ def test_extract_from_rolling_state_format(): def test_extract_returns_none_when_missing(): assert _extract_version_from_name("just_a_folder_name") is None - - -# ---------- _filter_patch_compatible ---------------------------------------- - - -def test_filter_keeps_same_patch(): - folders = [("id1", "0.1.114#alice@example.com")] - assert _filter_patch_compatible(folders, current_version="0.1.114") == folders - - -def test_filter_keeps_different_patch_same_minor(): - folders = [("id1", "0.1.114#alice@example.com")] - assert _filter_patch_compatible(folders, current_version="0.1.200") == folders - - -def test_filter_drops_minor_diff(): - folders = [ - ("id1", "0.1.114#alice@example.com"), - ("id2", "0.2.0#alice@example.com"), - ] - assert _filter_patch_compatible(folders, current_version="0.1.114") == [ - ("id1", "0.1.114#alice@example.com") - ] - - -def test_filter_drops_major_diff(): - folders = [ - ("id1", "0.1.114#alice@example.com"), - ("id2", "1.0.0#alice@example.com"), - ] - assert _filter_patch_compatible(folders, current_version="0.1.114") == [ - ("id1", "0.1.114#alice@example.com") - ] - - -def test_filter_drops_names_without_a_version(): - folders = [ - ("id1", "0.1.114#alice@example.com"), - ("id2", "no_version_here"), - ] - assert _filter_patch_compatible(folders, current_version="0.1.114") == [ - ("id1", "0.1.114#alice@example.com") - ] - - -def test_filter_covers_all_four_folder_formats(): - """All four formats syft-client uses should match when major.minor align.""" - folders = [ - ("id1", "0.1.114#alice@example.com"), - ("id2", "syft_datasite#0.1.115#alice@example.com#inbox#bob@example.com"), - ("id3", "alice@example.com-0.1.116-checkpoints"), - ("id4", "alice@example.com-0.1.117-rolling-state"), - ] - kept = _filter_patch_compatible(folders, current_version="0.1.200") - assert {fid for fid, _ in kept} == {"id1", "id2", "id3", "id4"} - - -def test_filter_returns_empty_for_bad_current_version(): - folders = [("id1", "0.1.114#alice@example.com")] - assert _filter_patch_compatible(folders, current_version="garbage") == []