docs: async def handle_control_event(self, message):
          """Handle an event from the Channels layer.
  
          Channels layer callback, do not call directly.
          """
          cmd = message[WorkerProtocol.COMMAND]
          logger.debug(__("Manager worker got channel command '{}'.", cmd))
  
          # Prepare settings for use; Django overlaid by state overlaid by
          # anything immediate in the current packet.
          immediates = {}
          if cmd == WorkerProtocol.COMMUNICATE:
              immediates = message.get(WorkerProtocol.COMMUNICATE_SETTINGS, {}) or {}
          override = self.state.settings_override or {}
          override.update(immediates)
          self.settings_actual = self._marshal_settings()
          self.settings_actual.update(override)
  
          if cmd == WorkerProtocol.COMMUNICATE:
              try:
                  await database_sync_to_async(self._data_scan)(**message[WorkerProtocol.COMMUNICATE_EXTRA])
              except Exception:
                  logger.exception("Unknown error occured while processing communicate control command.")
                  raise
              finally:
                  await self.sync_counter.dec('communicate')
  
          elif cmd == WorkerProtocol.FINISH:
              try:
                  data_id = message[WorkerProtocol.DATA_ID]
                  data_location = DataLocation.objects.get(data__id=data_id)
                  if not getattr(settings, 'FLOW_MANAGER_KEEP_DATA', False):
                      try:
                          def handle_error(func, path, exc_info):
                              """Handle permission errors while removing data directories."""
                              if isinstance(exc_info[1], PermissionError):
                                  os.chmod(path, 0o700)
                                  shutil.rmtree(path)
  
                          # Remove secrets directory, but leave the rest of the runtime directory
                          # intact. Runtime directory will be removed during data purge, when the
                          # data object is removed.
                          secrets_dir = os.path.join(
                              self._get_per_data_dir('RUNTIME_DIR', data_location.subpath),
                              ExecutorFiles.SECRETS_DIR
                          )
                          shutil.rmtree(secrets_dir, onerror=handle_error)
                      except OSError:
                          logger.exception("Manager exception while removing data runtime directory.")
  
                  if message[WorkerProtocol.FINISH_SPAWNED]:
                      await database_sync_to_async(self._data_scan)(**message[WorkerProtocol.FINISH_COMMUNICATE_EXTRA])
              except Exception:
                  logger.exception(
                      "Unknown error occured while processing finish control command.",
                      extra={'data_id': data_id}
                  )
                  raise
              finally:
                  await self.sync_counter.dec('executor')
  
          elif cmd == WorkerProtocol.ABORT:
              await self.sync_counter.dec('executor')
  
          else:
              logger.error(__("Ignoring unknown manager control command '{}'.", cmd))