Brighter Feedback

[iancooper]

@iancooper please could you help us validate more issues we're anticipating in situations with:

• Number of Brighter instances running reaching ~50
• Rate of messages being produced upwards of 250 per second

Our current understanding

During PostAsync the following happens:

• Message is deposited into the outbox
• Message is attempted to be cleared from the outbox
• Also CheckOutstandingMessages

and during the sweeper process:

• Get all outstanding messages (in our case with Kafka by default at least 5 seconds old)
• Attempt to clear all of them from the outbox

Potential Issue

Race condition we're concerned with are:

• Tick 0: Brighter process 1 begins PostAsync on message A
• Tick 1 to 50 million: Brighter process 1 still awaiting success response for message A
• Tick 50 million and 1 (5s later): Brighter process 2 begins sweeping and picks up message A as outstanding
• ...: Message A is published twice

Since an attempt to clear the outbox will be made by both sweepers which each Brighter process will have and as part of every PostAsync this would lead to a number of duplicate messages being published proportionate to the rate of messages being published and the number of sweeper processes competing against each other.

Potential Solution

Assuming our understanding is correct, we're thinking it may make sense to implement a kind of 'visibility timeout'? So during DynamoDbOutbox.OutstandingMessages we would query for:

KeyExpression => TopicShard = :v_TopicShard and CreatedTime < :v_CreatedTime
FilterExpression => VisibilityTimeout > :v_VisibilityTimeout

Where we also update the visibility timeout again inside of DynamoDbOutbox.OutstandingMessages to be now + {ConfiguredVisibilityTimeout}.

Wouldn't be perfect as there's no consistent reads for the GSI but would help mitigate those race conditions.

What do you think?

[iancooper]

For the time being, is it possible and would it make sense to effectively 'turn off' any sort of outbox with Brighter in it's current state? We understand we'd potentially be trading off throughput but gaining our preference to not unknowingly drop messages.

from @jtsalva moved from #2810

[iancooper]

During PostAsync the following happens:

Message is deposited into the outbox
Message is attempted to be cleared from the outbox
Also CheckOutstandingMessages

Sending a message in Brighter involves two steps:

• DepositPost: Add your outgoing message to the outbox (in memory if you have not supplied one), participating in a transaction if you use configure a transaction provider for use with the outbox that you DI into your handler. Returns a list of added messages.
• ClearAsync: Dispatches a list of messages using your configured messaging producer (and message mappers).

PostAsync does DepositPost followed by ClearAsync on the list of saved messages, and does not participate in your transaction.

and during the sweeper process:

Get all outstanding messages (in our case with Kafka by default at least 5 seconds old)
Attempt to clear all of them from the outbox

Yes

Race condition we're concerned with are:

Tick 0: Brighter process 1 begins PostAsync on message A
Tick 1 to 50 million: Brighter process 1 still awaiting success response for message A
Tick 50 million and 1 (5s later): Brighter process 2 begins sweeping and picks up message A as outstanding
...: Message A is published twice

Yep. In distributed systems in order for you to have guaranteed delivery, you must accept the possibility of duplicates. This is why we refer to guaranteed, at least once. The only way to get guaranteed, exactly once is to de-duplicate at the consumer side. Kafka has some support for that, but we also offer an Inbox which will allow you to reject a message that you have already successfully handled.

This is a known limitation of distributed systems.

Assuming our understanding is correct, we're thinking it may make sense to implement a kind of 'visibility timeout'?
So you could do this, to effectively limit how long before you forward a failed message. It might have some utility outside of the duplicate use case i.e. at a certain point this message would have no meaning. If it is just duplication, I would tend to rely on an Inbox, because the race condition is usually rare, but we could raise a request for a feature that says: outbox messages time out.

@preardon Any thoughts on this idea? Something you have seen?

[jtsalva]

Reading more into Kafka's de-duplication 'The idempotent delivery option guarantees that resending a message will not result in duplicate entries in the log, and that log order is maintained. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer with every message.' - my understanding is that wouldn't help in the (probably rare) case where multiple sweepers pick up the same set of messages at the same time from different Brighter instances as they'll have different producer IDs

[jtsalva]

Oh I see what you mean now by 'their transactional scope does not encompass any transactional scope that you have' yep

[iancooper]

where multiple sweepers pick up the same set of messages
Use just the one sweeper, and restart on crash

[jtsalva]

Use just the one sweeper, and restart on crash

How can we ensure only 1 sweeper is running out of the many Brighter processes?

[iancooper]

How can we ensure only 1 sweeper is running out of the many Brighter processes?
An in-process sweeper only uses an in-memory buffer, so you can use many of them.

An out-of-process sweeper, we recommend that you only have one that you deploy as a separate item. There is an example with that.

[iancooper]

@jtsalva ^^^