Fixed race-condition for streambroker.

[s3rius]

This PR fixes#123 .

I tried multiple approaches and found this one working.

Details

importasyncioimportredis.asyncioasaioredisSTREAM, GROUP="taskiq_bug", "taskiq"asyncdefsetup():
r=awaitaioredis.from_url("redis://localhost:6379")
awaitr.delete(STREAM)
awaitr.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
msg_id=awaitr.xadd(STREAM, {"data": b"test"})
awaitr.xreadgroup(GROUP, "worker-seed", {STREAM: ">"}, count=1)
awaitr.aclose()
returnmsg_idasyncdefworker(name: str):
conn=awaitaioredis.from_url("redis://localhost:6379")
lock_key=f"autoclaim:{GROUP}:{STREAM}"pipe=conn.pipeline()
lock=pipe.lock(lock_key)
awaitlock.acquire()
pipe.xautoclaim(
STREAM,
GROUP,
name,
min_idle_time=100,
count=10,
)
awaitlock.release()
res=awaitpipe.execute()
ifres[1][1]:
msgs=res[1][1]
print(name, "got", msgs[0][0])
awaitconn.aclose()
asyncdefrun_once():
awaitsetup()
awaitasyncio.sleep(0.2)
awaitasyncio.gather(
worker("worker-0"),
worker("worker-1"),
)
asyncdefmain():
for_inrange(25):
awaitrun_once()
if__name__=="__main__":
asyncio.run(main())

Output:

worker-0 got b'1782170479725-0'
worker-1 got b'1782170479929-0'
worker-0 got b'1782170480133-0'
worker-0 got b'1782170480337-0'
worker-0 got b'1782170480540-0'
worker-0 got b'1782170480744-0'
worker-0 got b'1782170480947-0'
worker-0 got b'1782170481150-0'
worker-0 got b'1782170481354-0'
worker-0 got b'1782170481557-0'
worker-0 got b'1782170481760-0'
worker-0 got b'1782170481963-0'
worker-0 got b'1782170482166-0'
worker-0 got b'1782170482369-0'
worker-1 got b'1782170482572-0'
worker-1 got b'1782170482775-0'
worker-0 got b'1782170482978-0'
worker-0 got b'1782170483180-0'
worker-0 got b'1782170483383-0'
worker-0 got b'1782170483586-0'
worker-0 got b'1782170483790-0'
worker-1 got b'1782170483993-0'
worker-0 got b'1782170484197-0'
worker-0 got b'1782170484402-0'
worker-0 got b'1782170484605-0'

Here's redis monitor output:

1782169992.799451 [0 172.19.0.1:37304] "MULTI"
1782169992.799458 [0 172.19.0.1:37304] "SET" "autoclaim:taskiq:taskiq_bug" "f1015f306e8f11f1ac1603adb8b6c235" "NX"
1782169992.799472 [0 172.19.0.1:37304] "XAUTOCLAIM" "taskiq_bug" "taskiq" "worker-0" "100" "0-0" "COUNT" "10"
1782169992.799474 [0 172.19.0.1:37304] "EVALSHA" "c3f8721cbb97f72bc19e972846bd7aaf91901658" "1" "autoclaim:taskiq:taskiq_bug" "f1015f306e8f11f1ac1603adb8b6c235"
1782169992.799487 [0 lua] "get" "autoclaim:taskiq:taskiq_bug"
1782169992.799491 [0 lua] "del" "autoclaim:taskiq:taskiq_bug"
1782169992.799494 [0 172.19.0.1:37304] "EXEC"
1782169992.799503 [0 172.19.0.1:37316] "MULTI"
1782169992.799506 [0 172.19.0.1:37316] "SET" "autoclaim:taskiq:taskiq_bug" "f1017c546e8f11f1ac1603adb8b6c235" "NX"
1782169992.799509 [0 172.19.0.1:37316] "XAUTOCLAIM" "taskiq_bug" "taskiq" "worker-1" "100" "0-0" "COUNT" "10"
1782169992.799510 [0 172.19.0.1:37316] "EVALSHA" "c3f8721cbb97f72bc19e972846bd7aaf91901658" "1" "autoclaim:taskiq:taskiq_bug" "f1017c546e8f11f1ac1603adb8b6c235"
1782169992.799513 [0 lua] "get" "autoclaim:taskiq:taskiq_bug"
1782169992.799514 [0 lua] "del" "autoclaim:taskiq:taskiq_bug"
1782169992.799515 [0 172.19.0.1:37316] "EXEC"

It seems to do exactly what is needed.

[codecov]

Codecov Report

❌ Patch coverage is 0% with 11 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.18%. Comparing base (090a00a) to head (5d3d514).
⚠️ Report is 36 commits behind head on main.

Files with missing lines	Patch %	Lines
taskiq_redis/redis_broker.py	0.00%	11 Missing ⚠️

Additional details and impacted files

@@ Coverage Diff @@## main #126 +/- ##
==========================================
+ Coverage 91.70% 92.18% +0.48% 
==========================================
Files 7 8 +1 Lines 434 704 +270 ==========================================
+ Hits 398 649 +251 - Misses 36 55 +19 

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.