Actor is never stopped when a dispatch update changes its merge identity #310

What happened?

ActorDispatcher finds the actor for a dispatch through dispatch_identity(dispatch), which is MergeStrategy.identity() when a merge strategy is used. MergeByTypeTarget puts the dispatch target in the identity, and target can be updated on a running dispatch.

When that happens, the running-state event carries the new identity, so _start_actor() creates a second actor for the new target, and the actor for the old target is never stopped. When the dispatch ends, the stop event also uses the new identity, so only the new actor is stopped. The old one keeps acting on the old target until the process restarts.

Minimal reproduction:

import asyncio
from types import SimpleNamespace

from frequenz.channels import Broadcast
from frequenz.dispatch import MergeByTypeTarget
from frequenz.dispatch._actor_dispatcher import ActorDispatcher
from frequenz.sdk.actor import Actor

actors = []

class Dummy(Actor):
    def __init__(self, info):
        super().__init__(name=f"dummy(target={info.target})")
        actors.append(self)

    async def _run(self):
        await asyncio.Event().wait()

async def factory(info, rx):
    return Dummy(info)

def dispatch(target, started):
    return SimpleNamespace(id=1, type="T", dry_run=False, started=started, target=target, payload={})

async def main():
    chan = Broadcast(name="running")
    tx = chan.new_sender()
    ad = ActorDispatcher(factory, chan.new_receiver(), dispatch_identity=MergeByTypeTarget().identity)
    ad.start()
    await tx.send(dispatch(["BATTERY"], started=True))    # dispatch starts
    await asyncio.sleep(0.1)
    await tx.send(dispatch(["INVERTER"], started=True))   # same dispatch, target updated
    await asyncio.sleep(0.1)
    await tx.send(dispatch(["INVERTER"], started=False))  # dispatch ends
    await asyncio.sleep(0.1)
    for a in actors:
        print(a.name, "running" if a.is_running else "stopped")
    await ad.stop()

asyncio.run(main())

Output:

dummy(target=['BATTERY']) running
dummy(target=['INVERTER']) stopped

MergeByType isn't affected, because its identity only uses type and dry_run, which the API doesn't allow updating. A custom strategy whose identity uses any other updatable field is affected the same way.

What did you expect instead?

When an update moves a running dispatch to a different identity, the actor for the old identity should be stopped (unless another running dispatch still maps to it), and the actor for the new identity started or updated.

Affected version(s)

v1.0.4

Affected part(s)

Actor (part:actor)

Extra information

ActorDispatcher only sees the updated Dispatch, so it can't know the previous identity. It probably needs to remember the identity each dispatch ID currently maps to, and treat a change as a stop for the old identity plus a start for the new one, respecting the merge strategy's filter() so other dispatches that keep the old identity running aren't affected.