Design & implement system that reduces storage pressure and provides client feedback when PUSH is throttled

[vladsud]

This is fork from #6596. That issue will continue to track handling of the nack with retryAfter, while this issue will track the following issues:

• The lack of feedback to client when PUSH is throttled due to flushing ops. Currently, 99% of that is due to new client joining, as ops flushing is coalesced with new summary commitment, and thus has direct way to provide client feedback in the form of nack.
• Optional - reduction or complete elimination of a need to flush ops by PUSH to storage on client joining.
• Optional - exposing a way to fetch ops from PUSH (redis).

The key goal should be to push throttling up the stack, such that if the issue is - too high ops rate, then client has enough information to learn about it and possibly adjust its behavior. While long term we may want to even disallow users to make changes, for now we will focus consuming this signal in summarization workflow, i.e. make sure client does not keep piling uncommitted summaries while PUSH has trouble and thus creates positive feedback loop.

The discussion of signals in #6596 is directly related to this problem that we fork into this issue.

[vladsud]

Here is the summary of discussion between @GaryWilber, @marcmasmsft & me:

1. We commit to a plan for SPO fetch ops API to be single source of truth for ops, and thus fetch ops (on as needed bases) from PUSH / redis
   • the details needs to be designed. It's likely that client would need to provide some hint for this flow to be engaged, both as a staging mechanism, but also as a way to reduce extra work on the server (later might be indirect, for example we should disambiguate cases where client asked for specific range, vs. client proactively asks for op tail with no specific expectation on how many ops are available).
   • There is some uncertainty here due to 3 entities involved (and service to service calls), so we need to test feasibility of it.
   • Ops commit while submitting summary needs more discussion. It's possible that we will stay with existing flow (i.e. push model, not pull) which would require client to ask push to flush ops before client can proceed with submitting summary in single commit flow.
2. Marcelo to work on design doc, publish it, and figure out execution plan.
3. This will take substantial time to evolve, so if we need some shorter mechanism to solve immediate scalability problems, then we should address it independently.

Temp solution in this space should be staged in a way where service can retire such temp capability easily, in other words - client (and whole system) should behave correctly when such new capability is missing or become missing in the future.

[vladsud]

Here is a proposal from me on how to address this problem (or rather consideration points). I start with just flush() API as consideration, but look into getOps() type of API as well (it has very similar requirements RE implementation).

1. PUSH to implement flush() API (available via via socket messages) to flush some set (current?) of ops to SPO.
2. Client to provide indication in connection message if client supports flushing and will use it if server supports it (enlightened clients).
   • server would not flush ops on connection for such client, otherwise it keeps old behavior.
3. bto communicate back to client if it supports flushing
   • Client should keep existing behavior (not calling this API) if server does not support it, such that PUSH can revoke this API in the future without notice.
4. Client usage of flush:
   • not flushing on summary (at least for now with 2-phase summary commit), with assumption that ops flushing is bundled with summary upload by PUSH and nack is used for communicating response.
     - we can revisit this in the future.
   • Client will ask to flush ops on hitting gap in ops (after preforming first successful storage ops fetch call but not filling in the gap)
     • Providing feedback (success / failure) is optional from POV of client that tries to fetch missing ops (as such client can just keep asking SPO for ops for 30 seconds, though feedback would reduce frequency and make telemetry more actionable).
     • But it's not optional from POV of summarizer client who is blocked by this operation. So there is a need to provide that feedback (at least 429 / any other recoverable failures) to all clients.
     • instead of flush, we can use getOps() here with similar impact to client. In such case feedback communication is part of API. Perf characteristics are slightly different. It will result in fewer SPO calls, and improved latencies, so it definitely needs to be considered, but flush() is more versatile as it maybe used in the future as a pre-step for summary upload.
5. Benefits (assuming all enlightened clients):
   • PUSH does not need to flush ops on connection, substantially reducing number of SPO calls (in some situations, including expiring tokens), as well as provide recovery mechanism for fetching ops that were lost through pipeline.
6. Gap:
   • if we have a mix of clients, then enlightened clients need to know about anybody (clients, PUSH) doing flushes and hitting problems. But that's not enough - summarizer may be non-enlightened client and thus not aware why summaries are not coming through and thus continue to post summaries and make things worse (positive feedback loop).

With that, I think there are two possible practical workflows:

1. flush() and signaling to when PUSH has issues on any flush().
   • Pros:
     • has slightly better characteristics in terms of client awareness of what's going on in a mix of enlightened & non-enlightened clients (if summarizer is enlightened).
     • possibly reusable for other workflows
   • Cons: has more implementation nuances (signals need to be re-broadcast to any new clients that joins, signals can be lost, etc.)
2. getOps()
   • Pros:
     • More localized (it's just single client activity)
     • Better perf profile / latency
   • Cons: Mix of enlightened & non-enlightened clients will behave worse in terms of scaling down on throttling failures.

My choice would be to go with getOps() kind of API, which I'd guess will look like this:

• "getOps" message from client
• ops are returned through normal (socket) flow, maybe have some marker that they are result of prior getOps call (it may help client to assert that normal ops have to come sequentially).
• some message back to client indicating completion of the call (client can use it to understand if gap was filled in or not). If failures can occur, they are communicated on same message.

[anthony-murphy]

I wonder if this is all necessary.

Why does push need to flush on each join?

We support IDocumentDeltaConnection.initialMessages? Does this not include all unflushed ops? Isn't that its purpose? is it not sufficient?

Even with IDocumentDeltaConnection.initialMessages there is still the problem of feedback. There is a server to client feedback that pushing ops to storage is behind so we should hold off summarizing, and maybe in the future take more drastic action like going readonly. The other feedback is client to server, basically that it cannot find some op gap. If we only have to solve this, and not the above i wonder if other solutions are possible.

[vladsud]

Gary would have more precise description here, there is no guarantee on ordering / lack of gaps. I do not remember precise reasoning here, but it was something to do with fetching these ops being async and potentially some ops being added to redis after this call is satisfied, and such ops being not broadcast earlier to that (connecting) client. So they exists only in redis past the point of connection (from client perspective) and are not available to client in any other form.

[anthony-murphy]

I'm not super familiar with this layer of the code, but i remember it working like this. The driver connects to the socket, which quickly starts sending ops, which the client should keep. The server then sends connect document success which includes outstanding ops as of that point, but only needs to contain up to the point of when the client started receiving ops. Older ops come from the server.

It's possible push started flushing on each join, as it wasn't possible to reliably do the above. @GaryWilber and @tanviraumi is this a push only behavior? AFIK r11s writes these directly to mongo, so no need to flush anything. It seems other solutions are possible here, like tracking the first seq on each conn, and ensuring server side, that messages up to that seq are either flushed, or in initial messages.

also, @vladsud do we envision this all being encapsulated in the driver? Seems like it could be hard to encapsulate with the separation of delta stream and delta storage

[GaryWilber]

@anthony-murphy Yes it's a push only behavior.

For r11s ops are inserted into mongodb. When clients want to get ops it ends up checking mongodb. So there is a single source of truth for that data.

For push, ops are either in SPO or Push. And technically ops could be in 2 separate places within Push.
Clients can access SPO ops via the get ops api and clients get the Push Redis ops via initialMessages. But due to how we are broadcasting messages (we broadcast the op before inserting it into Redis), there is point in time when the op is technically inaccessible. The op could be in a kafka topic, waiting to be consumed and inserted into Redis. For docs with high op rates, clients are likely to not get all the ops when they connect due to this. This is why we always post ops to SPO when clients join.

I have prototyped the solution you described, where we track the join op seq# and send the missed ops to the client. However that solution cannot work for read-only connections because there is no join op for them, so I dropped working on that further because I think we need a more general solution for this.

One other solution is to insert ops into redis before broadcasting them. This would ensure clients joining will always get all the ops. However this would increase latency because we would now be waiting for that redis insert call before broadcasting - we didn't want that to be a thing.. but maybe it would actually be worth since it simplifies the system & removes this problem. I'm conflicted about it.

[GaryWilber]

@vladsud Here's what the "get_ops" and "flush_ops" websocket apis look like right now.

Both apis have a request and a response and take in a client id & and object as params. Both objects contain a client provided nonce.

Get ops - only returns ops from Push's Redis

If all ops in the inclusive range based on the from & to params was found, code is 200. The ops are returned in the messages property.

If no ops were found, code is 404

If only some ops were found in the range, code is 206

Flush ops:

It returns a 200 if it flushed ops and includes the last persisted sequence number.
If no ops are flushed, it returns a 204.

In error cases the apis will return a code & retryAfter property (if available).

I could make the get_ops api broadcast the ops normally instead of making them only be returned in "get_ops_response" if you would prefer that. I would probably rename the event to "request_ops" instead of "get_ops" in that case.

[anthony-murphy]

@GaryWilber I don't know enough about how the server code is structured, but ideally, we can know the first op sent to any client, and then the server knows it needs to send the gap from that op down to the last flushed op in initial messages which is part of connect document success. This should be agnostic to write vs read.

[GaryWilber]

@anthony-murphy That is a great idea. Originally the frontends were "dumb". They did not track what it was broadcasting to clients - they simply broadcasted whatever it was told and never inspected the payloads. This is how r11s works too.

Due to some issues seen in the scale tests, I recently had to make the frontends smarter. They now parse the ops being broadcasted to clients. It tracks the sequence numbers per-document in order to detect issues (op gaps). Now that we have that capability, it should be possible to extend that logic further for per-client tracking too.

@vladsud Would a client recover correctly if this situation occurs:

1. Client gets initialMessages [1, 2, 3].
2. Server broadcasts 6.
3. Server broadcasts 7.
4. One second later, the server broadcasts 4, 5 to that client.

I think that was supported at some point but I forget if that's still okay or not. Today the per-document op gap recovery logic will not do that. But adding per-client op gap recovery logic may result in that sequence of events.

[vladsud]

Yes, the client will recover just fine.
It will make a call to SPO on step 2. It will cancel it (today that means - it will stop retrying for nothing infinitely) once # 4 occurs.
We can improve here and let it wait with a call a bit (in anticipation that gap would be filled in).

But I'm not sure this solves it, or rather the sequence you described is the one where we have issues. I think it's more like that:

1. Client initiates connection, that triggers request to redis
2. Client starts receiving ops 5, 6
3. # 1 completes and client receives connect_document_success message with initialMessages with 1, 2.
4. Client keeps receiving ops 7, 8, ...

I believe that's what happens (can confirm if needed RE observed behavior from client side) and it happens because of asynchrony in the system - ops 3, 4 are on the way to redis but were missed by call in # 1.

So if PUSH wants to fill in gaps, in needs to detect gap between very first op it sent and last op in initialMessages and attempt to fill it in.

Please correct me if I got it wrong.

[GaryWilber]

That's right. I think both scenarios can happen (the one I listed and yours). The main issue is what you described:

So if PUSH wants to fill in gaps, in needs to detect gap between very first op it sent and last op in initialMessages and attempt to fill it in.

Push should be able to detect that and fetch the gap ops and send them (per-client). I am going to try testing this in spdf this week.

[anthony-murphy]

awesome, it would be great it this works, and doesn't necessitate any protocol changes. This still leaves the issue of client feedback in general if push is throttled for whatever reason. I'm really curious if @tanviraumi or someone he knows had to solve a similar issue around their throttling design. Does it just used NACKs? or are there other mechanisms?

[vladsud]

I briefly talked to Gary and suggested authoring a document describing current behavior of PUSH (in respect to interactions to redis, kafka, etc., as well as document and client state tracking and processes involved around filling in ops gaps).

As for next steps, client would need to make a change to not immediately hit storage on op gap as PUSH will fill in a gap in next 0-20ms (with this new design), but ops coming out of order today result in immediate request to storage, which will be unnecessary with these changes. We already have enough telemetry to notice when we do unneeded storage calls, as well as cases where client can't fill in op gap in reasonable amount of time (30 seconds), so PUSH changes in this space could be validated by observing client telemetry.
The staging will look like this:

1. PUSH implements logic to fill in the gaps
2. Client implements delaying calling storage when gap is detected, in anticipation of PUSH filling in the gap in next N ms.
3. We validate via telemetry that client never does unneeded requests to storage (i.e. if client does request ops from storage to fill in the gap, then ops are always actually in storage). This tells us that # 1 is actually working and that threshold we chose is reasonable (it's probabilistic, so the goal of # 2 is reduce, but not completely eliminate calls to storage).
4. PUSH stops flushing ops to storage on client connection.
   • if there is a way to flight this change, we should, and validate it in our stress tests (likely by client having local change that provides some hint on connection to enable that behavior)
   • Validation here includes also ensuring that client never fails due to not being able to find ops anywhere for 30 seconds.

[vladsud]

Based on discussion at PUSH review meeting:
PUSH has a lot of logic to resolve various issues with reliability of op delivery and ordering guarantees. And even after set of latest fixes, ops are not ordered all the time, resulting in client needing # 2 from above.

It feels like we need to move further to one of the extremes:

1. PUSH continues on this path and provides strong ordering guarantees, ensuring that all ops are delivered in order at all times, including initial ops on connection (this is almost the case, but not guaranteed, ops can come out of order).
2. We remove most of the logic that helps with above from PUSH and instead force client (driver) to fetch ops from PUSH through getOps type of API (latest maybe moved to be brokered by SPO).
   • Solution is described at the beginning of this issue.
   • The biggest con of that approach is additional network latency. So feasibility of that path depends on how often it happens.

The main telemetry event that is telling is fluid:telemetry:DeltaManager:enqueueMessages event (reason & previousReason properties).

Query:

Office_Fluid_FluidRuntime_Performance
| where Data_eventName == "fluid:telemetry:DeltaManager:enqueueMessages"
| summarize count(), countif(Data_initialGap > 0), countif(Data_gap > 0), countif(Data_duplicate > 0) by Data_eventName, Data_reason, Data_previousReason
| order by count_

First, scale - appropriate set (limiting only to 0.4x loader versions as that's when this event was added) corresponds to

• 205K container sessions
• 563K (re)connections.

Terms mean:

• ReconnectOps means processing initial ops on some reconnect (not first connection for this container).
• InitialOps - same as above, but very first connection in container.
• DocumentOpen_pending means we fetched ops from storage on document open, but there was some gap or duplciate
• opHandler - op is coming from socket
• reason (one of the above) - where op came from when we detected gap or duplicate ops
• previousReason - what was the source (reason) of previously successfully processed op (before gap/duplication occured)

What this data tells us:

1. Ranges do not have that many gaps. I.e. there are cases where initial ops as observed by DeltaManager would have a gap due to establishing early op handler and potential for ops to trickle in while we wait for "connect_document_success" message to bring initial ops - such early ops are merged into initial ops set by driver. So number of gaps for reason == "InitialOps" or reason =="ReconnectOps" is at 300.
2. There are a lot of duplicates in initial ops (ReconnectOps) and ops coming from web socket (on previous connection), i.e. there are a ton of fast reconnects where initialOps bring same content that client saw before (over 30K cases)
3. Most interesting is initialGap property when reason == "opHandler", i.e. when we consume an op from socket and realize there is a gap. Here are the top offenders:

previousReason       count
opHandler	     10066
ReconnectOps	     6736
DocumentOpen_fetch   751

So in 10K cases we do see a gap in op stream itself (that's 2% of connections). I believe this is the issue with lost redis connection by front-end.
And in 6K cases (1.2% of connections) we see a gap between first op and initial ops on (presumably) same connection.

[vladsud]

And here is Gary's description of current system:

To recap the 2 things causing the gaps:

1. the fact that Push broadcasts to clients before inserting the op into redis. this results in ops are sometimes not accessible to newly joining clients
2. redis pubsub issues, so frontends sometimes lose ops

And the workarounds:

1. Push posting ops to spo whenver a client joins. This results in the client asking SPO for ops after joining and it gets the successfully
2. Added extra logic to detect these gaps, fetch the ops from redis, and broadcast them to the client.

With my change in SPDF, both are now fixed with the same logic.

1. The frontend detects gaps on a per document level and sends the ops to the clients
2. The frontend detects gaps on a per client level and sends the op to the client (the gap ops may be sent up to 5 seconds later after detection, so the client needs to handle non-sequential ops)

[vladsud]

@GaryWilber, some observations based on my prototyping (in priority order):

1. "If no ops are flushed, it returns a 204.". This is not very useful as it does not tell client the latest sequence number that SPO has. So client has no ability to make decision on what to do next (in situation where it needs to ensure that certain op made it to SPO in order to proceed with summarizaiton).
2. It is very useful for IConnected response to have supportedFeatures property listing all optional features like "api_flush_ops" & "api_get_ops" such that client can make decision on whether to use them, and whether to assert if it does not get proper response from server when such features are supported (i.e. defense in depth). For example, client for most part ensures that it will not have overlapping ops requests, so it could assert that responses indeed come back before next request shows up. Currently I can't do it as I do not know if particular PUSH box supports that feature.
3. Payloads do not have enough context on multiplexed socket for clients to understand if given response is applicable to given document. Similar, client can't tell if it's response for its request unless it tracks all nonces. In both cases it's possible to format nonce in a way where it has that data (i.e. it is formed in format of documentId-clientId-randomId, but it would be great for responses in both cases to contain documentId & clientId.

[GaryWilber]

@vladsud Great observations.

1. If no ops are flushed, it would hopefully mean that the latest op the client processed is already in SPO. However we know that ops are broadcasted before being inserted into Redis so that may not be the case.
   This is a bit tricky to solve because Push does not explicitly track the last persisted op to SPO. It is tracked by the deli lambda but it's not exposed to clients and it's not exposed in a way that's easy for me to grab during the flush_ops call. Is this a blocker for the single stage post summary work?

2. That does sound useful. For returning supportedFeatures, would this work:
   export type FluidSupportedFeatures = "api_get_ops" | "api_flush_ops"
   and add supportedFeatures?: FluidSupportedFeatures[]; to IConnected.

3. I will add tenantId, documentId & clientId to the responses.

[vladsud]

For # 1, I'll go for now with assumption that "all ops are flushed", but if we see SPO rejecting summary due to lack of ops in such cases, we might need to revisit it.

I'd prefer to go with connected.supportedFeatures = { api_get_ops: true, api_flush_ops: true };
Mostly to allow more extensible format (i.e. maybe some future features would want to communicate level of support, or some other properties).

[vladsud]

@GaryWilber, need more insights / feedback:

1. I'm getting 429s in our local stress tests in flush_ops_response. I have not looked at response, but I assume we do not have retryAfter there. We need it.
2. I'm also getting a lot of 409s. Why? There are no other file activities from client side (flush & summary upload are done in sequence by same client, and there are no images in stress test, so there are really no other activities from client that modify file). Is this because PUSH initiated flushes are async RE client initiated flushes? Shouldn't they be a single queue with no concurrency, and thus no way to hit 409?

[GaryWilber]

@vladsud

1. retryAfter will be there. Let me know if it's not.
2. Yes it's because the flushes are async between the two. The flushes clients run can happen at the same time as the ones Push run. The client initiated flushes do not run through the same queue system so they can happen concurrently.
   I looked at the 409s. From what I see, every single conflict was from Push posting ops due to client joins at the same time. So once we stop posting ops when clients joins, you should see a large reduction in 409s

[vladsud]

Everything covered in this issue has been implemented, including get_ops & flush_ops flows.
Next step - we should start re-evaluating places where we added complexity on PUSH side that are no longer required with get_ops flow.
Specifically:

1. We can deprecate initialMessages flow - see Deprecate IConnected.initialMessages #7132
2. PUSH can stop flushing ops to SPO on client connection
3. We should evaluate any other places where PUSH can be simplified.

@GaryWilber - if you have any PBIs tracking any future work in this domain, it would be great to list them.
Otherwise I'm, not aware of any work on client in this domain, so closing it as fixed.