fix(datastore): Use a UnicastSubject instead of a ReplaySubject

[manueliglesias]

• PR title and description conform to Pull Request guidelines.

Issue #, if available:
Closes: #2217

Description of changes:
Use a UnicastSubject instead of a ReplaySubject.

Address OOM problems related to unbounded, ever growing queues for subscription messages on long running apps

How did you test these changes?
Hammered an API with mutations for a sustained period of 30 minutes and benchmarked on an emulator, look at how nice memory usage looks with the change.

After: (Using UnicastSubject)

Documentation update required?

• No
• Yes (Please include a PR link for the documentation update)

General Checklist

• Added Unit Tests
• Added Integration Tests
• Security oriented best practices and standards are followed (e.g. using input sanitization, principle of least privilege, etc)

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[joekiller]

Looks like a nice fix

[joekiller]

Histrionic fun context that might be another bug. When this memory leak due to inbound mutation/create/delete models persisting together bit me, i tried to work around it and only one of two attempts were successful.

First I tried tried to use the decorators to say no/disable subscribe however the result was the datastore online init would fail. The failure to init with subscription disabled is (or was) a bug as there is code indicating the state should be okay because outbound mutations and sync would still work. So that didn't work.

My second trick was to override the subscription vtls to just reply unauthorized which the datastore accepted.

So yay for this fix, no more hacks to prevent the leak.

Ps, I also made sync always reply unauthorized because the app is literally push/append only operations. Making sync say unauthorized was fastest startup with many data models (16+).

[joekiller]

stating here that this only fixes half of what #2217 proposed, the SubscriptionProcessor problem.

This doesn't close out the need to have predictable memory management in MutationOutbox.

[joekiller]

Nice work @manueliglesias