Test case: not all messages from the sse server are received by sse clients

[IgorGubin]

Description

Detailed description with example here:
https://github.com/IgorGubin/SseExample

Reproduction Steps

1. publish client and server applications
2. deploy the client's sse console application, in a folder for example C:\MS\SSE\client
3. run the server application in visual studio
4. start two cmd.exe processes and in these processes run the client's console application C:\MS\SSE\client\Client.exe one after another
5. sooner or later both clients will stop at the waiting step because the following will happen

Expected behavior

Each client's console application receives all server event messages, selects those from them by eventId that relate to its session, processes them and waits for the next events.

Actual behavior

At some point only one of the clients receives messages for itself and for the other client, while the other client does not receive the messages it expects and remains in the waiting state forever. And vice versa. This leads both clients to the blocking state.

Regression?

No response

Known Workarounds

No response

Configuration

No response

Other information

No response

[MihaZupan]

Looking at your repo there's a decent bit of custom logic going on.
Can you please create a minimal project that still reproduces this issue?

My first guess would be that this is either:

1. A server issue where the server logic isn't reliably broadcasting events to all subscribers
2. Connection issue - e.g. one of the client connections got closed and the reading loop around sse didn't realize it because of lack of cancellation.

[IgorGubin]

Michael, Hello!

Thanks for your promptness!

I use EventId as session identifier. Is it even correct to do this?
When the client starts, it generates a Session Id, which does not change until the end of its work.

Regarding simplification - I understand. I have already partially simplified the real project by replacing work with real files with imitation of work with them, leaving only the step of waiting for the file to be ready.
Of course, I can simplify the project even more by throwing out the receipt of the session identifier by the server by hardcoding a couple of SessionId values ​​for a couple of clients and passing the SessionId via a parameter when they are launched.
And instead of simulating file processing on the server, generate file maps and change their states by timer.
I will need some time for this.

If you continue to work with what exists now, then I suggest paying attention to lines like these in the client logs:

>>> Sessionid: "6b46ac89af924dd1a5a2a06452d9e984"  - got data [EventId: b9e3d68cbffd460ca51574b6e28c089f; EventType: StateInfo]: Id: "09d8ea9c27e74ffb9a816c5111f74109"; State: InProcessing; 

when Sessionid is not equal to EventId.

And in principle, it can receive messages with EventId that differ from its SessionId by ignoring them, but then the other client should also receive them, and should have a message like this:

>>> Sessionid: "b9e3d68cbffd460ca51574b6e28c089f"  - got data [EventId: b9e3d68cbffd460ca51574b6e28c089f; EventType: StateInfo]: Id: "09d8ea9c27e74ffb9a816c5111f74109"; State: InProcessing; 

for the same file "09d8ea9c27e74ffb9a816c5111f74109".

And one more note - starting processes from one process installation folder is not necessary. The problem remains even if each of clients is started from a separate installation folder. I think it will be even more convenient during testing, when each client would keep its own logs, instead of dealing with one huge log file.

[IgorGubin]

Michael, Hello!

Here simplified example from main branch
https://github.com/IgorGubin/SseExample/tree/simplify

I am ready to answer on any questions.

[MihaZupan]

Miha*

Looking at your server code, you're dequeueing updates from your controller
https://github.com/IgorGubin/SseExample/blob/0fc6f65d20d2dc0fadb4b6debe64a06d070c33a1/Server/Controllers/ApiController.cs#L67-L75

If you have multiple readers in that controllers (your two clients), both will be dequieing from the same queue. By definition only one of them will see the update.

The way your server controller is written you'll also use a full CPU core for every client that subscribes to events.
Consider using Channels instead of a busy polling loop. E.g. your controller could look something like

var channel = Channel.CreateBounded<string>(50);

Conveyor.Subscribers.TryAdd(channel, 0);
try
{
    await foreach (var update in channel.Reader.ReadAllAsync(context.RequestAborted))
    {
        await WriteEventAsync(context, update);
    }
}
finally
{
    Conveyor.Subscribers.TryRemove(channel);
}

and the producer side has a

public static ConcurrentDictionary<Channel<string>, byte> Subscribers { get; }

where you then loop over subscribers and write to all of them.

[IgorGubin]

Michael, Hello!

Thanks a lot for the suggestion to use Chanel.

I want to clarify some points:
1)

Looking at your server code, you're dequeueing updates from your controller

Deletion occurs not in the controller queue, but in the state change history queue in each of FileCard object. Object FileCard is created for each uploaded on server file and stored in the static card catalog Conveyor.TotalFileCardCatalog as it item.
I think your example in actually must be a bit more complicated.
But I got the idea.

2. I would also like to understand what is happening under the hood of the SSE connection.
   I understood the following, please correct me if I am wrong
   From the examples and explanations that I was able to find on the Internet:
   when the HTTP client makes a call

using var stream = await client.GetStreamAsync(srvStatesUrl);

to the action of controller on the server

[HttpGet("states")]
public async Task StatesStream ... 

A persistent connection is created and a stream is assigned to it.
Then if the server doing any recording by using Response.WriteAsync,

async Task WriteEvent<T>(HttpResponse response, string? id, T data)
{
    var eventName = typeof(T).Name;

    await response.WriteAsync($"id: {id}\n");
    await response.WriteAsync($"event: {eventName}\n");
    await response.WriteAsync($"data: ");
    await JsonSerializer.SerializeAsync(response.Body, data);
    await response.WriteAsync($"\n\n");
    await response.Body.FlushAsync();
}

under the hood it performed broadband broadcast for all clients.
But that is exactly what is not happening.

If we follow your logic, context from under the hood should to provide an opportunity to get all the Response objects for each of the sse-clients, and next for each of them to call the method WriteAsync. Maybe I just don't know yet how to do it.

[MihaZupan]

It's Miha, not Michael*

Each call to your ASP.NET controller corresponds to a single client. There is no built-in broadcast in ASP.NET.
A call to response.WriteAsync will only send data to 1 client.
Each HttpClient.GetStreamAsync corresponds to one client in your StatesStream.

[IgorGubin]

Miha, Hello!

Thanks for your promptness!

I think I understood - I need to pass the session identifier in some http header and in the GET method that records in the sse-stream, do additional filtering of the data related to this session in conveyor.TotalFileCardCatalog, by Request.Headers. Or I will pass the Id parameter (sessionId) to the Get method.

By the way, does SSE technology provide any http header for transmitting SessionId?

I'll try this later.

[MihaZupan]

SSE provides a way for streaming messages to the client.
If you want to pass extra context from the client to the server producing said messages, you can use regular HTTP headers (or the query parameters, etc).

Closing this as a non-issue with the SSE logic from our end.