There's no reviewboard/broker/base/__init__.py file (exacerbated by us using namespaces = false in pyproject.toml's tool.setuptools.packages.find)

BaseBrokerBackend creates a self.logger but everything in here uses a module-level logger.

This logs the message but never resets the value. We'd then hit the assertion below in dev, or just crash in other ways in prod.

We should probably create with a more locked-down permissions mask.

I don't know if you've had a chance to look at the spec I put together on tasks, but I don't think we should use a result backend.
Instead, what I had planned out was a task model that could store (smaller) results directly, and larger things (like doc conversion or review bot) end up mapping to domain objects like reviews or file attachments.
If we end up actually using a result backend, this would need to be fixed to use "file://localhost{queue_path}" (no extra slash). Celery strips off 16 characters for "file://localhost". For absolute queue paths that currently ends up with a double leading slash, and if somehow the queue path is a relative path, it would end up adding an extra leading slash turning it into an (incorrect) absolute path.

This doesn't match the implementation/tests.
For send_task, the command name is in the headers (message_headers['task']) and the body is [args, kwargs, embed] as per Celery protocol 2.
For broadcast, the body is a {'method': ..., 'arguments': ...} dict.

This should probably protect with threading.Lock.
It would be nice to also listen to the siteconfig sync (or put in a TODO comment about that).

There's a lot of duplication in here.
It might be nice to have a _get_broker_settings(broker_id) helper which can validate and return the settings. That would be independently testable, and reusable by BaseBrokerSettingsForm.__init__, which currently does the same lookup but with no validation.

This is using a module-level logger instead of self.logger. Same for the other one just below.

Celery's ctor doesn't define a name arg, but it does take in **kwargs and silently ignore anything it doesn't know about.
I think this was intended to be main='reviewboard'

These should match the same order as they appear in the signature.

This is passing instance as a positional arg, but the first argument for KeyValueForm.__init__() is data. We need to explicitly pass instance=instance here.

Can we format as:

from djblets.registries.registry import (
    ALREADY_REGISTERED,
    ...
    UNREGISTER,
)

This is defining the entry point but the class doesn't inherit from EntryPointRegistry.
If we fix that we also need a yield from super().get_defaults() inside the get_defaults() implementation.

Our other registries use %(item)s. As-is this will render as:



"<class 'LocalBrokerBackend'>" is already a registered broker backend.

(with both quotes and repr). Alternatively just remove the quotes.

Same comment about %(item)s vs %(item)r

Should be reviewboard.broker.base.backends.BaseBrokerBackend

It looks like this is using an outdated name for get_celery_config. Same for other tests below.

Any time we open a file in text mode we should pass encoding=. When that's present I prefer to also be explicit with mode='r'.

The mode here only applies to the leaf directory. If we want all dirs to be set to that we need to call os.umask() first (although maybe it's fine for parent dirs to have the default perms).

Something got weird here: "If state had to be added to repaired"
Was that supposed to be "added or repaired"?

We should wrap this with with self._lock:.

get_broker_settings() already is typed as returning a dict. If that dict is empty, we then return a separate dict instead. Let's just return get_broker_settings() so we don't have to worry about aliasing issues.

message_files is already joined to queue_dir because of the. glob() call. We should be able to just do with open(message_file, ...): here.

Should sort before cryptography

This could fit on one line.

New files need to be added to coderef/index.rst

One thought I just had: the registry holds backend classes, but the backends themselves hold per-instance state (like locks and the cached Celery instance). shutdown() only means anything if it's called on the same instance that get_celery() populated.
Should we have the registry instead store instances? Or at least create a singleton accessor?

Maybe clarify that this is only on Linux systems?

This can go in the if TYPE_CHECKING block

This is pointing to some very old celery docs. Can we point to /en/stable/internals/protocol.html?

Do we want to raise an exception if this is empty? Base class sets to '', so we're currently silently hoping that subclasses override it.

This should include a "Raises" section with kombu.exceptions.OperationalError

This should include a "Raises" section with kombu.exceptions.OperationalError

We discussed it on the above issue but it never got fixed: can we get rid of or {} here?

This is operating somewhat differently to other stuff in the codebase which is similar:

This is fetching its own siteconfig object. A future "Message broker" page that's using these as subforms will end up having two separate siteconfig instances that get saved independently.
Additionally, this is the one of our KeyValueForms where the instance and the saved object are different. It's only working because the instance is identity aliased inside of siteconfig.settings.
It's inheriting KeyValueForm.get_key_value, so broker settings never see the siteconfig defaults layer, only field.initial. That's fine for now because the default is empty, but might bite us in the future.
save() is called without update_fields=('settings',), and no load_site_config() call after.

The closest analogs to how this form is operating are SearchBackendForm and the auth/SSO backends. Those solve these issues in somewhat different ways: for earch, the SearchSettingsForm handles all the siteconfig persistence. For auth/SSO, the subforms are their own SiteSettingsForm, which works because all the auth keys are flat instead of nested objects.
I think ideally we'd mirror what SearchBackendForm does, but at a minimum, instead of fetching our own siteconfig, we should inject it, and comment about how instance is reaching into siteconfig:
```python

def init(

    self,

    siteconfig: SiteConfiguration,

args,

    broker_cls: type[BaseBrokerBackend],

*kwargs,

) -> None:

    self.broker_cls = broker_cls

    self.siteconfig = siteconfig
super().__init__(
    *args,
    instance=broker_cls.get_broker_settings(),
    **kwargs)

def save_instance(self) -> None:

    # self.instance is the dict already stored within the siteconfig's

    # broker settings, so any mutations have already been applied. We just

    # need to persist them.

    self.siteconfig.save(update_fields=('settings',))

This is unused.

Typo: extra space between BaseBroker and Backend

Does this work if you run the test in isolation, or is it silently order-dependent on the other tests that are setting this key?
This method also never restores the key. How about:

old_settings = siteconfig.settings.pop('brokers', None)

if old_settings is not None:
    self.addCleanup(siteconfig.settings.__setitem__, 'brokers', old_settings)
else:
  self.addCleanup(siteconfig.settings.pop, 'brokers', None)

Do we want to pin this narrow, or do ">=5.6.3,<6"?