DuplicateIds Error When Mounting Toast Too Fast

[ADevAJ]

Have you checked closed issues? https://github.com/Textualize/textual/issues?q=is%3Aissue+is%3Aclosed

Please give a brief but clear explanation of the issue. If you can, include a complete working example that demonstrates the bug. Check it can run without modifications.

It will be helpful if you run the following command and paste the results:

textual diagnose

Feel free to add screenshots and / or videos. These can be very helpful!

[ADevAJ]

from textual import events, on, work
from textual.worker import Worker, get_current_worker
from textual.app import App, ComposeResult
from textual.widgets import Header, Footer

class DuplicateIDError_App(App):
    """Too many toasts too fast produces duplicate widget ID error"""

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Footer()

    @work(exclusive=False, thread=True)
    async def func1(self) -> None:
        worker = get_current_worker()
        if not worker.is_cancelled: self.notify("func1!")
    @work(exclusive=False, thread=True)
    async def func2(self) -> None:
        worker = get_current_worker()
        if not worker.is_cancelled: self.notify("func2!")
    @work(exclusive=False, thread=True)
    async def func3(self) -> None:
        worker = get_current_worker()
        if not worker.is_cancelled: self.notify("func3!")
    @work(exclusive=False, thread=True)
    async def func4(self) -> None:
        worker = get_current_worker()
        if not worker.is_cancelled: self.notify("func4!")
        
    def timers_mount(self, _timers : dict):
     for k in _timers:

      _interval = int(_timers[k])
      _func_str="self.app." + k
 
      timer_object = self.set_interval(_interval, eval(_func_str))

    def on_mount(self):
     func_names_and_intervals = {"func1":5, "func2":10,"func3":2,"func4":4}
     self.timers_mount(func_names_and_intervals)

if __name__ == "__main__":

    app = DuplicateIDError_App()
    app.title = "Duplicate ID Error Demo"
    app.run()

[ADevAJ]

program runs a few seconds, then returns exceptions as pictured below

[davep]

It looks to me like the issue here is that the code doesn't observe this warning from the documentation on threaded workers:

Most Textual functions are not thread-safe which means you will need to use call_from_thread to run them from a thread worker. An exception would be post_message which is thread-safe. If your worker needs to make multiple updates to the UI, it is a good idea to send custom messages and let the message handler update the state of the UI.

Calling notify directly from a threaded worker would be an example of something not to do. Your calls to show a notification should, as per the documentation, either be a post_message to the application where it calls notify, or alternatively something like this in the threaded worker code:

self.call_from_thread(self.notify, "func1")