Race condition when calling Publish directly after ServeHTTP

[strowk]

There is a problem with the current implementation of this lib, is that if server is implemented to create routine that calls ServeHTTP, then calls Publish, the result could be that message is ignored.

ServeHTTP calls Subscribe here:

go-sse/server.go

Lines 168 to 175 in e3ddbdf

	if err = s.provider.Subscribe(r.Context(), sub); err != nil {
	if l != nil {
	l.Error("sse: subscribe error", "error", err)
	}
	http.Error(w, err.Error(), http.StatusInternalServerError)
	return
	}

which by default would write subscription to channel:

go-sse/joe.go

Line 131 in e3ddbdf

case j.subscription <- subscription{done: done, Subscription: sub}:

which ideally should be processed and stored here:

go-sse/joe.go

Lines 262 to 277 in e3ddbdf

	case sub := <-j.subscription:
	var err error
	if replay != nil {
	err = tryReplay(sub.Subscription, &replay)
	}
	// NOTE(tmaxmax): We can't meaningfully handle replay panics in any way
	// other than disabling replay altogether. This ensures uptime
	// in the face of unexpected – returning the panic as an error
	// to the subscriber doesn't make sense, as it's probably not the subscriber's fault.
	if _, isPanic := err.(replayPanic); err != nil && !isPanic { //nolint:errorlint // it's our error
	sub.done <- err
	close(sub.done)
	} else {
	j.subscribers[sub.done] = sub.Subscription
	}

, before Publish caused list of subscribers to be checked:

go-sse/joe.go

Lines 242 to 243 in e3ddbdf

	for done, sub := range j.subscribers {
	if topicsIntersect(sub.Topics, msg.topics) {

In case if last two bits happened in the wrong order, the result would be data loss, i.e message that was published and is expected to be delivered, could be ignored by this library.

I believe there should be some sort of a way for server to signal whenever subscriber is stored, i.e some kind of channel should be written to after this:

go-sse/joe.go

Lines 275 to 277 in e3ddbdf

	} else {
	j.subscribers[sub.done] = sub.Subscription
	}

Ideally I'd expect some sort of Ready channel that caller can pass to Server and server should be writing in there after subscription was stored on the provider.

[tmaxmax]

hi there, thank you for opening this issue.

to be precise, this would be expected behavior. a "race condition" implies a bug and there is no bug here – it's just the non-deterministic nature of concurrent programs.

semantics aside, could you please provide a reproducible example of the issue and explain how it impacts your use case? generally it would be distinct, unrelated routines, which subscribe new connections and which publish new messages. it wouldn't be neither possible nor desired to try to synchronise the two.

routine that calls ServeHTTP, then calls Publish

the default ServeHTTP implementation is blocking, the same goroutine couldn't call Subscribe and Publish sequentially. it would have to be two separate goroutines – but then call order is not a well-defined concept anymore as the two goroutines operate concurrently. you'd have to use a synchronisation mechanism to re-establish order, which seems to be what you're suggesting to be added to go-sse. though, again, I can't see a use-case for this: the server can't determine when new clients will connect; a publish and a subscribe are usually two completely independent and uncorrelated events.

unless you want to publish a message to a specific client when the client connects. in that case #31 may help.

let me know exactly what you're trying to achieve and i will be able to assist you further.

[strowk]

you want to publish a message to a specific client when the client connect

Pretty much this yeah. Though I would probably be adding some time delay before doing this, so race would be side stepped..

The specifics are such that I am working on some messaging system that provides a delivery guarantee, so messages that were not confirmed would be sent (possibly again) when new connection is established. Or so was the goal at the time.