)]}'
{"/PATCHSET_LEVEL":[{"author":{"_account_id":6968,"name":"Christian Schwede","email":"cschwede@nvidia.com","username":"cschwede"},"change_message_id":"b082b4a462f49c2d00703c6c61ff72e5c29f5958","unresolved":false,"context_lines":[],"source_content_type":"","patch_set":1,"id":"b5b409eb_d977ab0d","updated":"2026-05-29 13:43:09.000000000","message":"Thx for your first contribution to Swift!\n\nA few notes:\n1. The existing tests are not covering this implementation properly. Do you have tests that need to be included in the patch?\n2. pool.waitall() runs without any limit - is this intended?\n3. sync_row_concurrency needs to be included in etc/container-server.conf-sample and doc/source/overview_container_sync.rst","commit_id":"d86001471180a71a744ea586af6aa3ed056a3729"},{"author":{"_account_id":38405,"name":"LEE JUYEONG","display_name":"LEE JUYEONG","email":"ijy102727@gmail.com","username":"ale8ander"},"change_message_id":"5b5e3bca5b40549387e442275488d7f5385d579c","unresolved":false,"context_lines":[],"source_content_type":"","patch_set":1,"id":"435e416b_fe969232","in_reply_to":"b5b409eb_d977ab0d","updated":"2026-06-05 14:45:39.000000000","message":"Thank for your feedback.\n\nI uploaded patch set 2.\n\nAnswer to few notes:\n    1. I’m still not very familiar with test_sync.py yet, so I’m refining the tests             as I go. I’d appreciate any feedback on the logic as well.\n\n    2. Yes, this is intended. pool.waitall() waits until synchronization of all rows currently being processed by the pool has completed. This follows the same behavior as the existing implementation, which also waits indefinitely for the synchronization of a row to complete without imposing a timeout.\n\n    3. I added those files in patch set 2.","commit_id":"d86001471180a71a744ea586af6aa3ed056a3729"},{"author":{"_account_id":38496,"name":"Andressa Cabistani","display_name":"Andressa","email":"acabistani@gmail.com","username":"andressadotpy","status":"I\u0027m a Software Engineer at Red Hat and I love Open Source and connect with people! Feel free to DM through IRC, I\u0027ll be delighted to chat"},"change_message_id":"686c9d31d429421d913ca599e38e699adc7d8b3f","unresolved":false,"context_lines":[],"source_content_type":"","patch_set":2,"id":"1069c513_07fa54e5","updated":"2026-07-07 12:37:25.000000000","message":"First of all, thank you for your contribution! It\u0027s really nice what you are all doing. I reviewed this patch carefully and I can see you put significant effort into improving container_sync performance. The idea of using parallelization to speed up container sync is excellent and the implementation shows good understanding of `ContextPool` and Swift\u0027s concurrency patterns.\n\nI tested the patch in SAIO and wrote unit tests to verify the behaviour. I found an important issue with how sync_point advancement works that it needs to be fixed before merging this.\n\nThe issue is:\n`sync_point` advances immediately when a row is **spawned** to the worker pool:\n\n```python\npool.spawn(self.container_sync_row, row, ...)  # Returns immediately\nsync_point1 \u003d row[\u0027ROWID\u0027] # Advances BEFORE sync completes\nbroker.set_x_container_sync_points(sync_point1, None)\n```\n\nWhy this is an issue:\nContainer sync relies on sync_point for crash recovery. The sync_point should only advance past rows that have actually been synced successfully.\n\nScenario example:\n1. Rows 100-107 are spawned to the worker pool\n2. sync_point1 \u003d 107 is written to the database\n3. Row 103\u0027s HTTP request fails (destination server returns 500 error)\n4. Container-sync daemon crashes (power failure, OOM, operator kills process)\n5. On restart, daemon resumes from sync_point1 \u003d 107\n6. Result: Row 103 is never synced and never retried → permanent data loss\n\nWith Claude\u0027s help, I wrote a test to validate this issue in a SAIO machine and it will also help you to understand the problem.\n\n```python\n#!/usr/bin/env python3\n\"\"\"\nimport os\nimport sys\nimport unittest\nfrom unittest import mock\nimport time\n\n# Add Swift to path\nsys.path.insert(0, os.path.join(os.path.dirname(__file__), \u0027../../../\u0027))\n\nfrom swift.container import sync\nfrom swift.common.utils import Timestamp\nfrom swift.common.exceptions import ClientException\nfrom swift.common.storage_policy import StoragePolicy\nfrom test.unit import patch_policies\n\n\nclass FakeRing(object):\n    def __init__(self):\n        self.devs \u003d [{\u0027ip\u0027: \u002710.0.0.%s\u0027 % x, \u0027port\u0027: 1000 + x, \u0027device\u0027: \u0027sda\u0027}\n                     for x in range(3)]\n\n    def get_nodes(self, account, container\u003dNone, obj\u003dNone):\n        return 1, list(self.devs)\n\n\nclass FakeContainerBroker(object):\n    def __init__(self, path, metadata\u003dNone, info\u003dNone, deleted\u003dFalse,\n                 items_since\u003dNone):\n        self.db_file \u003d path\n        self.db_dir \u003d os.path.dirname(path)\n        self.metadata \u003d metadata if metadata else {}\n        self.info \u003d info if info else {}\n        self.deleted \u003d deleted\n        self.items_since \u003d items_since if items_since else []\n        self.sync_point1 \u003d -1\n        self.sync_point2 \u003d -1\n        self.sync_point_updates \u003d []  # Track all updates\n\n    def get_max_row(self):\n        return 1\n\n    def get_info(self):\n        return self.info\n\n    def is_deleted(self):\n        return self.deleted\n\n    def get_items_since(self, sync_point, limit):\n        if sync_point \u003c 0:\n            sync_point \u003d 0\n        # Return items after sync_point\n        result \u003d []\n        for item in self.items_since:\n            if item[\u0027ROWID\u0027] \u003e sync_point:\n                result.append(item)\n                if len(result) \u003e\u003d limit:\n                    break\n        return result\n\n    def set_x_container_sync_points(self, sync_point1, sync_point2):\n        # Track the update\n        self.sync_point_updates.append((sync_point1, sync_point2))\n        if sync_point1 is not None:\n            self.sync_point1 \u003d sync_point1\n        if sync_point2 is not None:\n            self.sync_point2 \u003d sync_point2\n\n\n@patch_policies([StoragePolicy(0, \u0027zero\u0027, True, object_ring\u003dFakeRing())])\nclass TestSyncPointAdvancementSafety(unittest.TestCase):\n\n    def test_parallel_sync_point_advancement_safety(self):\n        \"\"\"\n        Verify sync_point only advances for completed syncs, not spawned ones.\n\n        Scenario:\n        - Queue 5 rows for parallel sync (ROWIDs 1-5)\n        - Row 3 fails (simulated 500 error)\n        - Row 5 is slow (delayed response)\n        - Verify sync_point only advances to row 2 (highest contiguous success)\n\n        Expected with patch 990630: FAILS (sync_point advances to 5)\n        Expected with fix: PASSES (sync_point advances to 2)\n        \"\"\"\n        print(\"\\n\" + \"\u003d\"*70)\n        print(\"TEST: Parallel sync_point advancement safety\")\n        print(\"\u003d\"*70)\n\n        # Track which rows completed and in what order\n        completed \u003d []\n\n        def mock_sync_row(row, *args, **kwargs):\n            \"\"\"Mock container_sync_row that simulates failures and delays\"\"\"\n            rowid \u003d row[\u0027ROWID\u0027]\n            print(f\"  Worker processing ROWID {rowid}...\")\n\n            if rowid \u003d\u003d 3:\n                print(f\"  ROWID {rowid}: SIMULATING FAILURE (HTTP 500)\")\n                raise Exception(\"Simulated HTTP 500 error\")\n\n            if rowid \u003d\u003d 5:\n                print(f\"  ROWID {rowid}: SIMULATING SLOW RESPONSE (2s delay)\")\n                time.sleep(2)\n\n            completed.append(rowid)\n            print(f\"  ROWID {rowid}: SUCCESS (completed: {completed})\")\n            return True\n\n        # Create broker with 5 test rows\n        fcb \u003d FakeContainerBroker(\n            \u0027test.db\u0027,\n            info\u003d{\n                \u0027account\u0027: \u0027test_account\u0027,\n                \u0027container\u0027: \u0027test_container\u0027,\n                \u0027storage_policy_index\u0027: 0,\n                \u0027x_container_sync_point1\u0027: -1,\n                \u0027x_container_sync_point2\u0027: -1\n            },\n            metadata\u003d{\n                \u0027x-container-sync-to\u0027: (\u0027http://127.0.0.1/dest/container\u0027, 1),\n                \u0027x-container-sync-key\u0027: (\u0027testkey\u0027, 1)\n            },\n            items_since\u003d[\n                {\u0027ROWID\u0027: 1, \u0027name\u0027: \u0027obj1\u0027, \u0027deleted\u0027: 0, \u0027created_at\u0027: \u00271.0\u0027},\n                {\u0027ROWID\u0027: 2, \u0027name\u0027: \u0027obj2\u0027, \u0027deleted\u0027: 0, \u0027created_at\u0027: \u00272.0\u0027},\n                {\u0027ROWID\u0027: 3, \u0027name\u0027: \u0027obj3\u0027, \u0027deleted\u0027: 0, \u0027created_at\u0027: \u00273.0\u0027},\n                {\u0027ROWID\u0027: 4, \u0027name\u0027: \u0027obj4\u0027, \u0027deleted\u0027: 0, \u0027created_at\u0027: \u00274.0\u0027},\n                {\u0027ROWID\u0027: 5, \u0027name\u0027: \u0027obj5\u0027, \u0027deleted\u0027: 0, \u0027created_at\u0027: \u00275.0\u0027},\n            ]\n        )\n\n        # Create ContainerSync instance\n        cring \u003d FakeRing()\n        with mock.patch(\u0027swift.container.sync.InternalClient\u0027):\n            cs \u003d sync.ContainerSync({\u0027sync_row_concurrency\u0027: 8}, container_ring\u003dcring)\n\n        print(\"\\nInitial state:\")\n        print(f\"  sync_point1: {fcb.sync_point1}\")\n        print(f\"  Rows to sync: 5 (ROWIDs 1-5)\")\n        print(f\"  Configuration: sync_row_concurrency\u003d8\")\n        print()\n\n        # Run sync with mocked container_sync_row\n        with mock.patch(\u0027swift.container.sync.ContainerBroker\u0027, lambda p, logger\u003dNone: fcb), \\\n             mock.patch.object(cs, \u0027container_sync_row\u0027, mock_sync_row), \\\n             mock.patch(\u0027swift.container.sync.put_object\u0027, lambda *x, **y: None), \\\n             mock.patch(\u0027swift.container.sync.delete_object\u0027, lambda *x, **y: None):\n\n            cs._myips \u003d [\u002710.0.0.0\u0027]\n            cs._myport \u003d 1000\n            cs.allowed_sync_hosts \u003d [\u0027127.0.0.1\u0027]\n\n            print(\"Running container_sync...\\n\")\n            try:\n                cs.container_sync(\u0027test.db\u0027)\n            except Exception as e:\n                print(f\"\\nException during sync (may be expected): {e}\\n\")\n\n        print(\"\\n\" + \"-\"*70)\n        print(\"RESULTS:\")\n        print(\"-\"*70)\n        print(f\"Completed rows: {completed}\")\n        print(f\"Final sync_point1: {fcb.sync_point1}\")\n        print(f\"sync_point updates: {fcb.sync_point_updates}\")\n        print()\n\n        # CRITICAL ASSERTION:\n        # With the BUG: sync_point1 will be 5 (advanced when spawned)\n        # With the FIX: sync_point1 should be 2 (only advanced for completed work)\n\n        print(\"EXPECTED BEHAVIOR:\")\n        print(\"  - Row 3 failed, so sync_point should NOT advance past row 2\")\n        print(\"  - Even though rows 4 and 5 were spawned, they should not\")\n        print(\"    advance sync_point because row 3 (earlier) failed\")\n        print(\"  - sync_point should be 2 (highest contiguous success)\")\n        print()\n\n        if fcb.sync_point1 \u003d\u003d 5:\n            print(\"❌ BUG DETECTED: sync_point1 \u003d 5\")\n            print(\"   This means sync_point advanced when work was SPAWNED,\")\n            print(\"   not when it COMPLETED. Row 3 will be lost on crash!\")\n            print()\n            self.fail(\"sync_point advanced for spawned work, not completed work\")\n\n        elif fcb.sync_point1 \u003d\u003d 2:\n            print(\"✅ CORRECT: sync_point1 \u003d 2\")\n            print(\"   sync_point only advanced for completed work.\")\n            print(\"   On restart, rows 3-5 will be retried (no data loss).\")\n            print()\n\n        else:\n            print(f\"⚠️  UNEXPECTED: sync_point1 \u003d {fcb.sync_point1}\")\n            print(f\"   Expected either 2 (correct) or 5 (bug)\")\n            print()\n\n        # Verify rows 3-5 would be retried on restart\n        next_rows \u003d fcb.get_items_since(fcb.sync_point1, 10)\n        next_rowids \u003d [r[\u0027ROWID\u0027] for r in next_rows]\n        print(f\"Rows that will be retried on restart: {next_rowids}\")\n\n        if fcb.sync_point1 \u003d\u003d 2:\n            self.assertIn(3, next_rowids, \"Row 3 should be retried\")\n            self.assertIn(4, next_rowids, \"Row 4 should be retried\")\n            self.assertIn(5, next_rowids, \"Row 5 should be retried\")\n            print(\"✅ All failed/pending rows will be retried\")\n\n        print(\"\\n\" + \"\u003d\"*70)\n        print(\"TEST COMPLETE\")\n        print(\"\u003d\"*70)\n\n\nif __name__ \u003d\u003d \u0027__main__\u0027:\n    # Run with verbose output\n    unittest.main(verbosity\u003d2)\n\n```\n\nSo you can see the difference, if you run this test on your patch branch you\u0027ll see if failing, but against master it will pass.\n\nThe test will show that sync_point advanced when work was **spawned** not when it **completed**. Row 3 will be lost on crash.\n\nSuggested fix, although I didn\u0027t test it locally: you need to track which rows complete successfully, then only advance sync_point past those rows.\n\n---\n\n## Other observations:\n\nQuestion - have you tested this in SAIO with actual container sync between two containers? It would be interesting to:\n1) Create a container with 100 objects\n2) Enable container sync with sync_row_concurrency\u003d8\n3) Monitor the logs to see parallel requests happening\n4) Verify all objects sync correctly\n\nA preference of my, even though it\u0027s not a requirement, would be to add a log message as well, so the operators can see the parallelization is working.\n\nRequirement - I would like for you to add more unit tests for this patch. A few of those should cover\n- sync_point only advances for completed syncs, not spawned ones\n- worker exceptions don\u0027t cause silent data loss\n- failed rows are retried on next container_sync run\n\n---\nSo I\u0027ll vote -1 for now because of the data-loss risk, once the sync_point issue is fixed and tests are added, I\u0027m happy to re-test it.\n\nIn case your need any help let me know!\nKeep up the great work!","commit_id":"68e31fdedd882e67325e3c1886b56dc5f4cadf81"},{"author":{"_account_id":38405,"name":"LEE JUYEONG","display_name":"LEE JUYEONG","email":"ijy102727@gmail.com","username":"ale8ander"},"change_message_id":"c89b9db247127030bfee2417987bdb628673b4cf","unresolved":false,"context_lines":[],"source_content_type":"","patch_set":2,"id":"0b7ae63d_92fa23a7","in_reply_to":"1069c513_07fa54e5","updated":"2026-07-25 12:02:37.000000000","message":"Hi, thank you for taking the time to review the patch and for sharing such a detailed test scenario.\n\nI revisited the `sync_point` advancement behavior by comparing both the existing implementation and this patch.\n\nFirst, I noticed that the test scenario you suggested assumes a code path where an exception raised while processing a row inside `container_sync_row()` propagates directly out of the function. However, in the normal container-sync failure path, `container_sync_row()` handles the failure internally and returns `False` instead of propagating an exception.\n\nTo verify this, I modified the scenario to match the normal failure path by replacing `raise Exception` with `return False`, and tested both the original implementation and this patch. The result was that the behavior of `sync_point1` advancement was identical in both cases.\n\nSpecifically, while processing rows inside the `while sync_stage_time \u003c stop_at` loop (`sync.py:426`), `sync_point1` continues to advance to the current row\u0027s `ROWID` regardless of whether the row sync succeeds. In this patch, `sync_point1` is still advanced after `pool.spawn(...)` at `sync.py:444` and `sync.py:445`. However, this is not new behavior introduced by the patch—the original implementation also advances `sync_point1` in the same loop.\n\nThe important point is that this mechanism does not permanently skip failed rows. During the next `container_sync()` execution, the `while time() \u003c stop_at and sync_point2 \u003c sync_point1` loop (`sync.py:389`) is executed first. Inside this loop, previously unprocessed rows are retrieved one at a time using `broker.get_items_since(sync_point2, 1)` (`sync.py:390`). If synchronizing a row fails, `container_sync_row(...)` returns `False` (`sync.py:403`), and the logic in `sync.py:406-412` ensures that the failed row remains eligible for retry in the next run.\n\nTo make this behavior explicit, I added a new unit test:\n\n`test_failed_rows_retried_on_next_container_sync_run`\n\nThis test verifies that a row which fails during one `container_sync()` execution is retried during the following execution.\n\n------\n\nRegarding the test cases you suggested, here is my understanding:\n\n### 1) `sync_point` only advances for completed syncs, not spawned ones\n\nThe current implementation does not behave this way. Inside the `while sync_stage_time \u003c stop_at` loop (`sync.py:426`), `sync_point1` continues to advance. However, failed rows are revisited during the next execution via the `while time() \u003c stop_at and sync_point2 \u003c sync_point1` loop (`sync.py:389`). Therefore, this behavior is not introduced by my patch; it is consistent with the existing algorithm.\n\n### 2) worker exceptions don\u0027t cause silent data loss\n\nIn the normal failure path, `container_sync_row()` returns `False` instead of propagating an exception. The failed row is then revisited during the next `container_sync()` execution through the `while time() \u003c stop_at and sync_point2 \u003c sync_point1` loop (`sync.py:389`).\n\n### 3) failed rows are retried on next `container_sync` run\n\nI addressed this by adding the new unit test described above.\n\n------\n\nAdditionally, I have already validated the patch in a SAIO environment.\n\nI gradually increased the number of objects synchronized between two containers and repeatedly verified the behavior, testing with approximately **10,000 to 300,000 objects**. During these tests, I enabled `sync_row_concurrency\u003d8` and confirmed through the logs that synchronization requests were executed concurrently. After synchronization completed, I also verified that all objects were successfully replicated to the destination container.\n\n------\n\nAccordingly, this update includes the new unit test `test_failed_rows_retried_on_next_container_sync_run`, which verifies that failed rows are retried during the next `container_sync()` execution.\n\nIf I\u0027ve misunderstood your concern or intended scenario, I\u0027d be happy to revisit it based on your expected behavior. Thank you again for your careful review.","commit_id":"68e31fdedd882e67325e3c1886b56dc5f4cadf81"},{"author":{"_account_id":38368,"name":"Christian Ohanaja","display_name":"Christian Ohanaja","email":"cohanaja@nvidia.com","username":"cohanaja"},"change_message_id":"98c754f00154959694eb5e934d94289a753fe72c","unresolved":false,"context_lines":[],"source_content_type":"","patch_set":3,"id":"99f0522c_37e55a58","updated":"2026-07-27 18:11:25.000000000","message":"Thanks for your contribution! The numbers in the attached blueprint look really swell and your work is appreciated.","commit_id":"4d1e78a49fc53550f65e37810d8653988811bcff"}],"swift/container/sync.py":[{"author":{"_account_id":38368,"name":"Christian Ohanaja","display_name":"Christian Ohanaja","email":"cohanaja@nvidia.com","username":"cohanaja"},"change_message_id":"98c754f00154959694eb5e934d94289a753fe72c","unresolved":true,"context_lines":[{"line_number":202,"context_line":"        swift.common.db.DB_PREALLOCATION \u003d \\"},{"line_number":203,"context_line":"            config_true_value(conf.get(\u0027db_preallocation\u0027, \u0027f\u0027))"},{"line_number":204,"context_line":"        self.conn_timeout \u003d float(conf.get(\u0027conn_timeout\u0027, 5))"},{"line_number":205,"context_line":"        self.sync_row_concurrency \u003d max("},{"line_number":206,"context_line":"            1, int(conf.get(\u0027sync_row_concurrency\u0027) or 8))"},{"line_number":207,"context_line":"        request_tries \u003d int(conf.get(\u0027request_tries\u0027) or 3)"},{"line_number":208,"context_line":""},{"line_number":209,"context_line":"        internal_client_conf_path \u003d conf.get(\u0027internal_client_conf_path\u0027)"}],"source_content_type":"text/x-python","patch_set":3,"id":"923143ac_ab1c954b","line":206,"range":{"start_line":205,"start_character":0,"end_line":206,"end_character":58},"updated":"2026-07-27 18:11:25.000000000","message":"nit: Good that we\u0027re setting concurrency by default, but I think we could probably expect the user not set the value \u003c1. `int(conf.get(\u0027sync_row_concurrency\u0027) or 8)` should suffice.","commit_id":"4d1e78a49fc53550f65e37810d8653988811bcff"}]}