Implementing Websockets without exhausting database connection pool?

[kamakazikamikaze]

I'm encountering an issue with serving multiple clients over websockets.

In this portion of the project, Emmett is serving an index page which opens two websocket connections. Each websocket is serving updates from a database on a timed interval. The intent is to essentially "live stream" data to two tables on a timed interval.

from asyncio import sleep
from datetime import datetime, timedelta

from emmett import App, websocket
from emmett.orm import Database, Model, Field
from emmett.tools import ServicePipe

# Application Config
app = App(__name__)
app.config.db.adapter = "postgres"
app.config.db.host = ...
app.config.db.user = ...
app.config.db.password = ...
app.config.db.database = ...
app.config.db.keep_alive_timeout = 300

# Model definitions
class Score(Model):
    name = Field().string()
    points = Field().int()
    recent = Field().datetime()

class Event(Model):
    message = Field().text()

db = Database(app, pool_size=4)
db.define_models(Score, Event)

api = app.module(__name__, name="api", url_prefix="api")
api.pipeline = [db.pipe, ServicePipe("json")]

@app.route("/", template="scoreboard.html")
async def index():
    return {}

@api.websocket("scoreboard")
async def scoreboard():
    await websocket.accept()
    while True:
        await websocket.send(
            {
                "data": [
                    {
                        "name": row.name,
                        "points": row.points,
                        "last_activity": row.recent.strftime("%Y-%m-%d %H:%M:%S")
                    }
                    for row in Scoreboard.all().select()
                ]
            }
        )
        await sleep(15)


@api.websocket("eventlog")
async def eventlog():
    await websocket.accept()
    while True:
        await websocket.send(
            {
                "data": [
                    {
                        "message": row.message,
                    }
                    for row in Event.where(lambda e: e.timestamp >= datetime.now() - timedelta(seconds=15)).select()
                ]
            }
        )
        await sleep(15)

With 4 pooled database connections, two users can load the index page just fine. Any additional users after that will get a 500 error for the websocket connections and the server will throw a TimeoutError.

From what I can gather from the documentation, it seems that the database pipe opens the connection when entering the context (i.e. @api.websocket(...)) and does not release it back to the pool automatically, meaning the connection is essentially "dedicated" for the lifetime of the context. This is probably ideal for a request context rather than a websocket one. Would removing the db.pipe from the module pipeline and wrapping each loop in a with db.connection(): statement be the proper approach?

(I'm currently testing this approach which has not triggered the same error, but I want to verify that this method is recommended with Emmett's underlying design.)

[gi0baro]

@kamakazikamikaze correct, by default the database Pipe opens a connection in the open step of the pipeline, and release it on the close step. This is true both for requests and websockets mainly for 2 reasons:

• historical reasons
• giving the same DX regardless of requests/websockets: if your code interact with the database having the database pipe in the pipeline it is always ensured you have a database connection opened.
  I agree with you that this is far from ideal in websockets scenarios, but at the same time there's no way for Emmett to understand when to acquire/release database connections based on your route code.

So yes, if you want to use database connections only when you need them in websockets, you should not use db.pipe on websockets routes and use async with db.connection() blocks around your code. Based on your snippet, I would rewrite it like this:

from emmett import now

# Common module, we set url prefix and JSON serialization
api_common = app.module(__name__, name="api_common", url_prefix="api")
api_common.pipeline = [ServicePipe("json")]

# Request based routes
api = api_common.module(__name__, name="api")
api.pipeline = [db.pipe]

# WS based routes
ws = api_common.module(__name__, name="ws")

@ws.websocket("scoreboard")
async def scoreboard():
    await websocket.accept()
    while True:
        async with db.connection():
            rows = Scoreboard.all().select()
        await websocket.send(
            {
                "data": [
                    {
                        "name": row.name,
                        "points": row.points,
                        "last_activity": row.recent.strftime("%Y-%m-%d %H:%M:%S")
                    }
                    for row in rows
                ]
            }
        )
        await sleep(15)


@ws.websocket("eventlog")
async def eventlog():
    await websocket.accept()
    while True:
        async with db.connection():
            rows = Event.where(lambda e: e.timestamp >= now().subtract(seconds=15)).select()
        await websocket.send(
            {
                "data": [
                    {
                        "message": row.message,
                    }
                    for row in rows
                ]
            }
        )
        await sleep(15)

Note: the above snippet releases db connections immediately after selecting records and before sending data; that's doable only if you don't have any rowattr/rowmethod that performs additional db operations on serialization.

[kamakazikamikaze]

Thanks for the clarification. I also couldn't find where in the documentation that serialization was noted but recalled seeing it somewhere. I'll try to move some of the websocket responses to auto-serialize.

Out of curiosity, can update_values be abused to do cross-table queries so that rowmethod doesn't have to be used?

For example:

class Connection(Model):
    user = Field().string()
    last_connection = Field().datetime()

class Activity(Model):
    # Documentation uses `request`, which leads me to believe that if you have an HTTP request
    # where an object is automatically pulled from the database you might be able to use it,
    # but I don't think we can do this within a websocket
    alive = Field().bool(update=lambda: Connection.where(lambda c: c.user == ...).select().first().last_connection <= datetime.now().subtract(seconds=15))
    # ...

[gi0baro]

Looks too over-complicated to recommend.
Why can't alive be a method of Connection?

[gi0baro]

Btw, in general to perform db ops on record changes involving other models is preferable to use before_save and after_save

[gi0baro]

Also, I just noticed you use Field().<type>() whether you should use Field.<type>(), you're constructing double field objects every-time with no reason.

[kamakazikamikaze]

Looks too over-complicated to recommend. Why can't alive be a method of Connection?

It can be. I'm just inquiring out of curiosity. The before_*/after_* methods are what I use but I wanted to know how far you could push some of the features.

Also, I just noticed you use Field().<type>() whether you should use Field.<type>(), you're constructing double field objects every-time with no reason.

Oh dang, I had to look at my original code and what do you know, most of the models I wrote do exactly that (Field().<type>()). Thanks for bringing it to my attention!