Broadcasting to multiple Sinatra streams during a request results in a FiberError. How do I send server-side events correctly?

[nogweii]

(This is all part of my personal project, Syncpad. Source is available under AGPL-3.0.)

I'm trying to set up a synchronized editor powered by Server-Side Events. I've got it to the point where it saves the document but then crashes when trying to broadcast an update to the browsers due to a FiberError:

13.52s     info: Async::HTTP::Protocol::HTTP2::Request: Incoming request: POST "/doc". [oid=0x618] [ec=0x620] [pid=958146] [2025-06-05 21:43:48 -0700]
               | -> POST /doc
               | {
               |   "headers": {
               |     "user-agent": "Mozilla/5.0 (X11; Linux x86_64; rv:138.0) Gecko/20100101 Firefox/138.0",
               |     "accept": [
               |       "*/*"
               |     ],
               |     "accept-language": [
               |       "en-US",
               |       "en;q=0.5"
               |     ],
               |     "accept-encoding": [
               |       "gzip",
               |       "deflate",
               |       "br",
               |       "zstd"
               |     ],
               |     "referer": "https://localhost:9292/",
               |     "content-type": "text/plain;charset=UTF-8",
               |     "origin": [
               |       "https://localhost:9292"
               |     ],
               |     "dnt": [
               |       "1"
               |     ],
               |     "sec-gpc": [
               |       "1"
               |     ],
               |     "sec-fetch-dest": [
               |       "empty"
               |     ],
               |     "sec-fetch-mode": [
               |       "cors"
               |     ],
               |     "sec-fetch-site": [
               |       "same-origin"
               |     ],
               |     "priority": [
               |       "u=4"
               |     ],
               |     "te": [
               |       "trailers"
               |     ]
               |   },
               |   "address": "#<Addrinfo: 127.0.0.1:42370 TCP>"
               | }
2025-06-05 21:43:48 - FiberError - attempt to yield on a not resumed fiber (FiberError):
	/home/colin/.local/share/rbenv/versions/3.4.1/lib/ruby/gems/3.4.0/gems/sinatra-4.1.1/lib/sinatra/base.rb:493:in 'block in Sinatra::Helpers::Stream#<<'
	/home/colin/.local/share/rbenv/versions/3.4.1/lib/ruby/gems/3.4.0/gems/sinatra-4.1.1/lib/sinatra/base.rb:461:in 'Sinatra::Helpers::Stream.schedule'
	/home/colin/.local/share/rbenv/versions/3.4.1/lib/ruby/gems/3.4.0/gems/sinatra-4.1.1/lib/sinatra/base.rb:493:in 'Sinatra::Helpers::Stream#<<'
	/home/colin/code/syncpad/server.rb:36:in 'block (2 levels) in <class:SyncPad>'
	/home/colin/.local/share/rbenv/versions/3.4.1/lib/ruby/3.4.0/set.rb:504:in 'Hash#each_key'
	/home/colin/.local/share/rbenv/versions/3.4.1/lib/ruby/3.4.0/set.rb:504:in 'Set#each'
	/home/colin/code/syncpad/server.rb:35:in 'block in <class:SyncPad>'

This comes from the following (trimmed) code:

server.rb

require "sinatra/base"

class SyncPad < Sinatra::Base
  set :connections, Set.new

  get "/doc" do
    content_type "text/event-stream"
    stream :keep_open do |out|
      if settings.connections.add?(out)
        out.callback { settings.connections.delete(out) }
      end
      sleep 1
    end
  end

  post "/doc" do
    request.body.rewind
    body = request.body.read
    # saves the request body off...

    # In theory, write to each browser that the document has been updated with the new contents
    settings.connections.each do |out|
      out << "event: DocUpdate\ndata: #{body}\n\n"
    end

    204
  end
end

I'm pretty sure I'm missing something, and would love some help pointing it out. How do I correctly send an update to the list of clients after the document has been updated?

[ioquatix]

The problem is this expression: @front.call(data.to_s). You are reaching from one request into another response stream and writing directly to it. It's unsafe and in this case is causing flow control issues within the scheduler. While we could solve that problem, it's better to use an output buffer like this:

  get "/doc" do
    content_type "text/event-stream"
    stream :keep_open do |out|
      queue = Thread::Queue.new

      if settings.connections.add?(queue)
        out.callback { settings.connections.delete(out) }
      end
      
      while item = queue.pop
        out << "event: DocUpdate\ndata: #{item}\n\n"
      end
    end
  end

  post "/doc" do
    request.body.rewind
    body = request.body.read
    settings.kv["doc"] = body
    settings.connections.each do |queue|
      queue << body
    end

    204
  end

This will decouple the flow control and life cycle and correctly serialise the output. Hope this example is sufficient - let me know if you need more help/details.

[nogweii]

Maybe I misunderstood your example, but it appears to only work for one client per connection. The events aren't broadcasted to other clients. The browser that sends the event sees it's own event (which is something I'll need to fix eventually, I expect) but no other browser sees that event.

Here's my current code:

# SPDX-FileCopyrightText: 2025 Nogweii <https://nogweii.net>
# SPDX-License-Identifier: AGPL-3.0-only

require "sinatra/base"
require "moneta"
require "json"

class SyncPad < Sinatra::Base
  set :static_cache_control, [:public, max_age: 300]
  set :kv, Moneta.new(:File, threadsafe: true, dir: "moneta")
  set :connections, Set.new

  get "/" do
    erb :index
  end

  get "/doc.txt" do
    content_type "text/plain"
    settings.kv["doc"]
  end

  get "/doc" do
    content_type "text/event-stream"
    stream :keep_open do |out|
      queue = Thread::Queue.new
      if settings.connections.add?(queue)
        out.callback { settings.connections.delete(queue) }
      end

      while (item = queue.pop)
        doc_event = {doc: item}
        out << "event: DocUpdate\ndata: #{doc_event.to_json}\n\n"
      end
    end
  end

  post "/doc" do
    request.body.rewind
    body = request.body.read
    settings.kv["doc"] = body
    settings.connections.each do |queue|
      queue << body
    end

    204
  end
end

My guess is it's because of .pop, and the queue is emptied before other clients get a chance to see it. Do I have that correct?

[ioquatix]

I've added a full working example here: https://github.com/socketry/falcon-examples/tree/main/streaming-sse-sinatra

Let me know how you get on.

[nogweii]

Huh, that example doesn't work for me. It loads, I see it POST to /broadcast, and nothing shows up in the other browser.

On a hunch, I ran bundle exec falcon serve -n 1 and it worked! With the default being my CPU core count, 16, was it always a 1/16 chance it'd work?

[ioquatix]

Ah right, yes, if you are doing it in-process, yes, 1/16th chance it would work. Otherwise, you need to use a shared backend pub/sub e.g. redis, etc.

[nogweii]

Is there a way to share the list? Or because Falcon actually launches distinct processes things wouldn't work out?

[ioquatix]

At this time, there is no native multi-process pub-sub, but it's something I'm thinking about in the future.