apoc.periodic.iterate yields corrupted data

[michaelncy]

Originally opened here: neo4j/apoc#823 (comment)

Expected Behavior

We're using an apoc.periodic.iterate query to delete relationships. Normally it yields all the output parameters correctly (for example updateStatistics is a map with the relevant fields).

Actual Behavior

We see where there's a deadlock in the query, the data that is yielded from the query (even though it succeeds after the deadlock) is corrupted, for example the updateStatistics is an int instead of a map.
We've also tried to run on our dev environment with yield only specific output parameters in order to avoid this issue, but we've still encountered the same problem, although seems like it's happening less frequently. Another error I saw on our dev env along side the original issue was: "ValueError('Invalid string length marker: 111')" originating from:

  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_async/work/result.py", line 605, in single
    await self._buffer(2)
  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_async/work/result.py", line 443, in _buffer
    async for record in self:
  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_async/work/result.py", line 393, in __aiter__
    await self._connection.fetch_message()
  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_async/io/_common.py", line 195, in inner
    await coroutine_func(*args, **kwargs)
  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_async/io/_bolt.py", line 991, in fetch_message
    tag, fields = await self.inbox.pop(
                  ^^^^^^^^^^^^^^^^^^^^^
  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_async/io/_common.py", line 81, in pop
    self._unpacker.unpack(hydration_hooks) for _ in range(size)
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/appuser/apps/test/.venv/lib/python3.12/site-packages/neo4j/_codec/packstream/v1/__init__.py", line 290, in unpack
    value, i = _rust_unpack(
               ^^^^^^^^^^^^^
ValueError: Invalid string length marker: 111

When using the regular neo4j driver the issue does not reproduce

How to Reproduce the Problem

Run an apoc.periodic.iterate query that deletes relationships with yielding all the output parameters.
In the same time run a query that creates relationships from one of the nodes in the first query to trigger a deadlock.

Simple Dataset

// First Query
CALL apoc.periodic.iterate("
    MATCH (:TEST)<-[r:REL]-(:BLA {a: $a}) RETURN id(r) AS id",
    "MATCH ()<-[r]-() WHERE id(r) = id DELETE r",
    {batchSize: 1000, parallel: false, retries: 5,  params: {a: $a}})

// Second Query - Pass a list of records to simulate a query that updates a batch of data
UNWIND $records AS record
MERGE (a:TEST)-[b:REL2]->(c:TEST2 {b: record})
ON CREATE SET a.test = "test"

Steps (Mandatory)

1. Run both queries at the same time to ensure a deadlock happens using the python client
2. Check the output from the first query

Screenshots (where it's possibile)

Specifications

Currently used versions

Versions

• OS: Neo4j Aura 2025.07
• Neo4j-Apoc: 2025.07.1
• Client: neo4j-python/5.26.0 Python-Rust/3.12.7-final-0 (linux)

[robsdedude]

Hi and thanks for reaching out.

This error is (assuming no bug in the driver/extensions) coming a protocol violation. I.e., the server is sending misformatted data.

I've checked the code-paths and compared the extensions vs the pure Python driver in this case, and they do indeed differ. The Python implementation when receiving a map just takes what ever value it gets and tries to use it as a dict key. The Rust implementation assumes that map keys are always strings (as the protocol specs mandate). So I assume when you inspect the map returned from the pure Python you'll somewhere find a dict with an non-string key in it, which neo4j should not be able to produce. I don't think this needs fixing in the Rust extensions. But I sill want to dig into it to make sure. And even then, it certainly needs fixing somewhere and it'd be a good start to figure out where that somewhere is.

I've tried to reproduce it (both, using 2025.07.1 running locally in docker as well as using Aura). Unfortunately, I couldn't.

Here's what I've tried:

Reproducer (not working)

import threading

import neo4j


URL = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")
DB = "neo4j"

RECORDS = [f"{i:07d}" for i in range(5_000)]


def setup(driver: neo4j.Driver) -> None:
    print("Wiping database...")
    with driver.session(database=DB) as session:
        # Clear the database
        session.run(
            "MATCH (n) "
            "CALL (n) { "
            "  DETACH DELETE n"
            "} IN 10 CONCURRENT TRANSACTIONS OF 1000 rows"
        ).consume()


    print("Populating database...")
    # Reproducer does not state what data exists in the database.
    # Is it empty?
    driver.execute_query(
        "UNWIND range(1, 500_000) AS i "
        "MERGE (s:TEST) "
        "MERGE (e:BLA {a: $a}) "
        "CREATE (s)<-[:REL]-(e) ",
        database_=DB,
        a="test",
    )


def worker1(driver: neo4j.Driver) -> None:
    print("worker1 starting...")
    with driver.session(database=DB) as session:
        result = session.run(
            'CALL apoc.periodic.iterate(\n'
            '  "MATCH (:TEST)<-[r:REL]-(:BLA {a: $a}) RETURN id(r) AS id",\n'
            '  "MATCH ()<-[r]-() WHERE id(r) = id DELETE r",\n'
            '  {batchSize: 1000, parallel: false, retries: 5,  params: {a: $a}}\n'
            ')',
            # $a is not defined in reproduction steps
            a="test",
        )
        eager_result = result.to_eager_result()
        print(f"worker1 done\n{eager_result}")


def worker2(driver: neo4j.Driver) -> None:
    print("worker2 starting...")
    with driver.session(database=DB) as session:
        result = session.run(
            "UNWIND $records AS record "
            "MERGE (a:TEST)-[b:REL2]->(c:TEST2 {b: record}) "
            'ON CREATE SET a.test = "test"',
            records=RECORDS,
        )
        eager_result = result.to_eager_result()
        print(f"worker2 done\n{eager_result}")


def main() -> None:
    with neo4j.GraphDatabase.driver(URL, auth=AUTH) as driver:
        setup(driver)

        print("Starting threads...")
        t1 = threading.Thread(target=worker1, args=(driver,), daemon=True)
        t2 = threading.Thread(target=worker2, args=(driver,), daemon=True)

        t1.start()
        t2.start()

        t1.join()
        t2.join()


if __name__ == "__main__":
    main()

Note how there are some assumptions I had to make because the reproduction steps in the original post are missing some details.

Could you kindly adjust the code such that it reproduces the issue for you when run against an empty DB and then share it with me?

[michaelncy]

Hi and thanks for reaching out.

This error is (assuming no bug in the driver/extensions) coming a protocol violation. I.e., the server is sending misformatted data.

I've checked the code-paths and compared the extensions vs the pure Python driver in this case, and they do indeed differ. The Python implementation when receiving a map just takes what ever value it gets and tries to use it as a dict key. The Rust implementation assumes that map keys are always strings (as the protocol specs mandate). So I assume when you inspect the map returned from the pure Python you'll somewhere find a dict with an non-string key in it, which neo4j should not be able to produce. I don't think this needs fixing in the Rust extensions. But I sill want to dig into it to make sure. And even then, it certainly needs fixing somewhere and it'd be a good start to figure out where that somewhere is.

I've tried to reproduce it (both, using 2025.07.1 running locally in docker as well as using Aura). Unfortunately, I couldn't.

Here's what I've tried:

Reproducer (not working)
import threading

import neo4j

URL = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")
DB = "neo4j"

RECORDS = [f"{i:07d}" for i in range(5_000)]

def setup(driver: neo4j.Driver) -> None:
print("Wiping database...")
with driver.session(database=DB) as session:
# Clear the database
session.run(
"MATCH (n) "
"CALL (n) { "
" DETACH DELETE n"
"} IN 10 CONCURRENT TRANSACTIONS OF 1000 rows"
).consume()

print("Populating database...")
# Reproducer does not state what data exists in the database.
# Is it empty?
driver.execute_query(
    "UNWIND range(1, 500_000) AS i "
    "MERGE (s:TEST) "
    "MERGE (e:BLA {a: $a}) "
    "CREATE (s)<-[:REL]-(e) ",
    database_=DB,
    a="test",
)

def worker1(driver: neo4j.Driver) -> None:
print("worker1 starting...")
with driver.session(database=DB) as session:
result = session.run(
'CALL apoc.periodic.iterate(\n'
' "MATCH (:TEST)<-[r:REL]-(:BLA {a: $a}) RETURN id(r) AS id",\n'
' "MATCH ()<-[r]-() WHERE id(r) = id DELETE r",\n'
' {batchSize: 1000, parallel: false, retries: 5, params: {a: $a}}\n'
')',
# $a is not defined in reproduction steps
a="test",
)
eager_result = result.to_eager_result()
print(f"worker1 done\n{eager_result}")

def worker2(driver: neo4j.Driver) -> None:
print("worker2 starting...")
with driver.session(database=DB) as session:
result = session.run(
"UNWIND $records AS record "
"MERGE (a:TEST)-[b:REL2]->(c:TEST2 {b: record}) "
'ON CREATE SET a.test = "test"',
records=RECORDS,
)
eager_result = result.to_eager_result()
print(f"worker2 done\n{eager_result}")

def main() -> None:
with neo4j.GraphDatabase.driver(URL, auth=AUTH) as driver:
setup(driver)

    print("Starting threads...")
    t1 = threading.Thread(target=worker1, args=(driver,), daemon=True)
    t2 = threading.Thread(target=worker2, args=(driver,), daemon=True)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

if name == "main":
main()
Note how there are some assumptions I had to make because the reproduction steps in the original post are missing some details.

Could you kindly adjust the code such that it reproduces the issue for you when run against an empty DB and then share it with me?

Hi @robsdedude,
Thanks for the thorough response.

The record we're getting from the apoc.iterate query from the driver looks like this, perhaps it can help with the debug:

I've managed to use this code on a clean DB to reproduce the issue:

import neo4j
import time
import threading

RECORDS = [i for i in range(1,50_001)]

async def worker1(driver: neo4j.Driver) -> None:
     print("worker1 starting...")
     async with driver.session(database="neo4j") as session:
         result = await session.run(
             'CALL apoc.periodic.iterate(\n'
             '  "MATCH ()-[r:REL]->(:TEST {a: $a}) RETURN r",\n'
             '  "DELETE r",\n'
             '  {batchSize: 10000, parallel: false, retries: 5,  params: {a: $a}}\n'
             ')',
             a="test",
         )
         eager_result = await result.to_eager_result()
         print(f"worker1 done\n{eager_result}")

async def worker2(driver: neo4j.Driver) -> None:
     print("worker2 starting...")
     async with driver.session(database="neo4j") as session:
         result = await session.run(
             "UNWIND $records AS record "
             "MERGE (a:BLA {id: record})"
             "MERGE (c:BLA2 {id: record})"
             "MERGE (a)-[b:REL2]->(c)"
             'ON CREATE SET b.test = "test"',
            #  "WITH *"
            #  "CALL apoc.util.sleep(1)",
             records=RECORDS,
         )
         eager_result = await result.to_eager_result()
         print(f"worker2 done\n{eager_result}")

async def main():
  driver = neo4j.AsyncGraphDatabase.driver("neo4j://neo4j-local",auth=("neo4j","password"))
  async with driver as driver:
    # Create constraints and indexes
    await driver.execute_query("CREATE CONSTRAINT IF NOT EXISTS FOR (b:BLA) REQUIRE (b.id) IS UNIQUE;")
    await driver.execute_query("CREATE CONSTRAINT IF NOT EXISTS FOR (b:BLA2) REQUIRE (b.id) IS UNIQUE;")
    await driver.execute_query("CREATE INDEX IF NOT EXISTS FOR (s:TEST) ON (s.a);")
  
    # Initialize empty db with data
    await driver.execute_query(
      "UNWIND range(1, 50_000) AS i "
      "MERGE (s:TEST {a: $a}) "
      "MERGE (e:BLA {id: i}) "
      "MERGE (e2:BLA2 {id: i})"
      "CREATE (e2)-[:REL]->(s)<-[:REL]-(e) ",
      database_="neo4j",
      a="test",
    )
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker1(driver))
        tg.create_task(worker2(driver))

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

This happens once in a few attempts for me, make sure you see a deadlock in the logs, as I think it's the key for the reproduction.
Thanks for looking into this!
Michael

[robsdedude]

Thank you very much for helping me to reproduce this. I had a hard time getting the same error, but ultimately managed to after changing the store format from block to aligned (just a note for the future, should this ever become relevant).

I've found a packstream decoding bug exclusive to the Rust extensions. The fix is linked and should be released in the coming days.

[michaelncy]

Thank you very much for helping me to reproduce this. I had a hard time getting the same error, but ultimately managed to after changing the store format from block to aligned (just a note for the future, should this ever become relevant).

I've found a packstream decoding bug exclusive to the Rust extensions. The fix is linked and should be released in the coming days.

Awesome!
Thanks for the quick response.

[robsdedude]

Fix released in version 5.28.2.1

[michaelncy]

Fix released in version 5.28.2.1

I checked the new version and it seems to have fixed the issue for us. Thanks for the quick fix!