tastyware/streaq:v6.0.0 [beta]

[Graeme22]

What's Changed

• Add Trio support (pending anyio support in coredis) by @Graeme22 in Add Trio support (pending anyio support in coredis) #80
  This is a large change, powered by simultaneous changes to coredis, the client used behind the scenes. This release is beta, pending the release of the next major coredis version. Here's a summary of the changes:
  • The library now works with both asyncio and Trio. You can change the backend by passing anyio_backend to workers:

    # asyncio is the default
    worker = Worker(anyio_backend="trio", anyio_kwargs={"restrict_keyboard_interrupt_to_checkpoints": True})

  • streaQ now uses structured concurrency everywhere for more predictability. If you haven't read Nathaniel Smith's article on structured concurrency, go read it now--it's a fantastic explanation of the motivations and an introduction to some of the concepts.
  • As a consequence of SC, async context managers for the worker are back! You'll need to use the context manager for any operation that involves communicating with Redis:

    async with worker:
        task = await sleeper.enqueue()
        print(await task.result(3))

    This means the recommended way of integrating with FastAPI has changed a bit:

    @asynccontextmanager
    async def app_lifespan(app: FastAPI) -> AsyncGenerator[None]:
        async with worker:
            yield
    
    app = FastAPI(lifespan=app_lifespan)

    See here for more details.
• Workers no longer attempt to re-enqueue tasks on shutdown--instead, we assume all shutdowns are unsafe, and allow Worker.idle_timeout to handle retrying. The default idle timeout has been decreased from 5 minutes to 1 minute to reflect this change.
• Redis scripts are now Redis functions in a single Lua library file. This means we now use FCALL instead of EVALSHA, which slightly increases performance in pipelines. This also means Redis versions less than 7.0 are no longer supported.
• Task abortion now handles delayed tasks better, aborting them right away instead of marking them and leaving them untouched
• TaskResult now has slightly better typing. Instead of accessing TaskResult.result which was either the task's output or any raised exception, TaskResult.result now only contains the task's output and errors if you access it when TaskResult.success is False. Exceptions are now contained in TaskResult.exception, and errors if you access it when TaskResult.success is True.

  async with worker:
      result = await task.result(3)
      if result.success:
          print(result.result)  # this is present only if success
      else:
          raise result.exception  # this is present only if not success

• Add date in logs by @nsteinmetz in Add date in logs #91
  Logs now contain date as well as time. Timezone can be customized as before.
• Dependencies bumped, now using dependabot (Bump actions/checkout from 4 to 5 #81, Bump actions/setup-python from 5 to 6 #86)
• Redis Sentinel now in test coverage everywhere a worker is used
• Tests now use a Docker compose cluster
• Added "Contributing" page to docs
• Benchmarks updated
• Web UI no longer uses alru_cache since it's not compatible with anyio

New Contributors

• @dependabot[bot] made their first contribution in Bump actions/checkout from 4 to 5 #81

Full Changelog: v5.2.2...v6.0.0

---

This discussion was created from the release tastyware/streaq:v6.0.0 [beta].

[darkydash]

Hi! I'm new to your library and maybe I'm missing a point. You say that streaq implements the structured concurrency which basically means "stop the main process and wait until all descendants finish".
Okay but streaq/arq/rq/whatever are the libraries to schedule stuff not simply in a child process but on the whole separate hosts. The main process may or may not become something it has to track (run a process and forget about it is a thing in some applications) now the second option is a bit limited altogether - and the usecase to await result some time in future is not clear from now (and to support this usecase we need to create a separate nursery each time making it some kind of duplicate to the raw await).
According to FastAPI example the whole app lifetime becomes the nursery of a single worker - and all the awaits around inside this huge nursery vanish the advantages of the structured concurrency (unless we create dozens of nurseries inside but why'd we have the outermost nursery then?). Let's assume we want our main app to be responsible for all the tasks it has spawned, fine, but how a nursery for a specific worker supposed to act in an app having multiple workers (watching after different queues for example)? Is it supposed we'd have to use async with Worker1() as w1, Worker2() as w2 or even AsyncExitStack? Then the question is - why shall I provide a list of each known workers if there could simply be one shared nursery?
How do I also act if I'm unaware of the worker being attached to a task? Why'd I need to go to the code and look after it explicitly if a task is already aware of the worker it's supposed to have initialized?

[darkydash]

I meant the main limitation here is that in 5.2.2 it was as easy to call a task (which, for example, is chosen to run in runtime) as await task.enqueue() without even worrying about the worker context. From 6.0.0 on I have to know exactly the worker which a task is bound for, even thought the task itself has this information. So if I choose the task in runtime (where you I need to have a huge context manager (or asyncexitstack) to count for every possible worker to be run on. So that's why I asked about the one common shared nursery in this context.
But after I studied the lib a bit more I see that to overcome this limit the async with task.worker: may be a thing - may it be considered to be a good practice?

[Graeme22]

I meant the main limitation here is that in 5.2.2 it was as easy to call a task (which, for example, is chosen to run in runtime) as await task.enqueue() without even worrying about the worker context. From 6.0.0 on I have to know exactly the worker which a task is bound for, even thought the task itself has this information. So if I choose the task in runtime (where you I need to have a huge context manager (or asyncexitstack) to count for every possible worker to be run on. So that's why I asked about the one common shared nursery in this context.

But after I studied the lib a bit more I see that to overcome this limit the async with task.worker: may be a thing - may it be considered to be a good practice?

5.2 was the only version that didn't have the context manager interestingly enough. You could just stay on that version if you want. The solution you proposed is valid, but note the context manager is not reentrant so you'll need to take that into account.

[darkydash]

5.2 was the only version that didn't have the context manager interestingly enough

I was lucky to find out about this lib right in the time lol

You could just stay on that version if you want.

I left arq because it feels abandoned so not something I'd want too...

The solution you proposed is valid, but note the context manager is not reentrant so you'll need to take that into account.

Oh, this will not be my use case but I think it could be a good idea to come up with some solution to this too because otherwise we need to recheck manually if we are already executing at some worker context :(

[Graeme22]

You can check if the context manager has already been entered with worker._initialized though this might lead to some race conditions if you're not careful, you'd probably need a lock as well.

The async exit stack for all the different workers would be a great solution as well. The context manager mostly just initializes the Redis connection pool and scripts, so if you do that once for each worker upon startup you should be good to go.

[darkydash]

Got it, thank you a lot! Will go with the exit stack approach then, seems without any drawbacks :P

[espdev]

Hi @Graeme22,

Are there any plans to release v6.0.0?

[Graeme22]

Yes, we're just waiting on coredis v6 release. Shouldn't be too long though, a week or two perhaps