You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Describe the bug
Our FastStream NATS implementation has seen a lot of error, since the start of using it:
Error in NATS: nats: 'Authentication Timeout'
Error in NATS: nats: connection closed
Error in NATS: Connection lost
Error setting key value: nats: no response from stream
Error in NATS: [Errno 104] Connection reset by peer
How to reproduce
Our NATSService, used by our regular FastAPI app:
And here is is our Dishka setup for the NATS Worker service which will handle NATS messages, the broker is used by the NATSService and the router by the FastAPI FastStream implementation:
importasyncioimportrandomfromtypingimportAnnotatedfromfastapiimportFastAPIfromfastapi.responsesimportORJSONResponsefromfaststream.natsimportDeliverPolicy, PullSubfromdishka.integrations.faststreamimportinject, FromDishkafromapi.authimportfirebase_initializationfromcommon.configimportget_configfromcommon.exceptions.handlersimportinit_exception_handlersfromcommon.middlewareimportinit_middlewarefromcommon.observabilityimportinit_observabilityfromcommon.services.nats_serviceimportMOMO_WORKER_DEFAULTfromcommon.services.user_serviceimportUserServicefromdiimportmake_containerfromdishka.integrations.faststreamimportsetup_dishkaassetup_faststream_iocfromdishka.integrations.fastapiimportsetup_dishkaassetup_fastapi_iocfromcommon.schemas.workerimportUserEmailSummarySendPayload## DI container#container, _, nats_router=make_container()
## FastStream setup#setup_faststream_ioc(
container,
nats_router,
finalize_container=False,
)
## FastAPI setup#_fastapi_app=FastAPI(
title="momo-worker",
lifespan=nats_router.lifespan_context,
docs_url=get_config().swagger_url(),
redoc_url=get_config().redoc_url(),
debug=get_config().DEBUG,
default_response_class=ORJSONResponse,
)
setup_fastapi_ioc(container, _fastapi_app)
init_observability("momo-worker", _fastapi_app)
init_exception_handlers(_fastapi_app)
init_middleware(_fastapi_app)
firebase_initialization()
# NOTE: subject: ENTITY.DOMAIN.ACTION# e.g. user.email.summary.send# e.g. organisation.feed.reset# e.g. organisation.event.created# e.g. organisation.event.deleted# e.g. organisation.event.started# e.g. organisation.event.ended# e.g. organisation.member.added## Handlers#@nats_router.subscriber("scheduled.bimonthly.user.email.summary.send",durable="bimonthly-user-email-summary-send",stream=MOMO_WORKER_DEFAULT,max_workers=10,deliver_policy=DeliverPolicy.ALL,pull_sub=PullSub(batch_size=10),)@injectasyncdefhandler(user_service: Annotated[UserService, FromDishka()]):
users=awaituser_service.get_all_users()
tasks= []
foruserinusers:
tasks.append(asyncio.create_task(nats_router.broker.publish(UserEmailSummarySendPayload(user_id=user.uid), "user.email.summary.send")))
awaitasyncio.gather(*tasks)
@nats_router.subscriber("user.email.summary.send",durable="user-email-summary-send",stream=MOMO_WORKER_DEFAULT,max_workers=10,deliver_policy=DeliverPolicy.ALL,pull_sub=PullSub(batch_size=10))@injectasyncdefhandler(msg: UserEmailSummarySendPayload, user_service: Annotated[UserService, FromDishka()]):
print(f"Sending email to {msg.user_id}")
try:
user=awaituser_service.get_user(msg.user_id)
exceptExceptionase:
passawaitasyncio.sleep(random.randint(1, 5))
returnmsg## Add routers#_fastapi_app.include_router(nats_router)
app=_fastapi_app
Expected behavior
We expect our NATSService and FastStream app to have constant connection with the NATS server (hosted on Scaleway), without the errors, and to automatically reconnect which is default behaviour.
Observed behavior
We observe connection errors, so does this suggest that the broker is not reconnecting?:
Error in NATS: nats: 'Authentication Timeout'
Error in NATS: nats: connection closed
Error in NATS: Connection lost
Error setting key value: nats: no response from stream
Error in NATS: [Errno 104] Connection reset by peer
Environment
Running FastStream 0.5.13 with CPython 3.11.6 on Darwin
Additional context
We have contacted Scaleway support.
They stated that resource limits are:
The limits are :
300Mb totals in stream/Kv
50 streams/Kv
50 consumers per stream
Which is more than enough, and we are not exceeding this.
They also stated that we must be sure to handle automatic reconnection. But to our understanding this is the default behaviour.
so are we doing something totally wrong with our implementation?
We are not always getting these errors. They seem to appear irregular.
The text was updated successfully, but these errors were encountered:
nats-py provides us with a connection recovery functional, but seems like recovery doesn't work if Authentication was failed. Also, it looks like Authentication fails on connection timeout too...
I need to dig into nats-py to find the reason
Describe the bug
Our FastStream NATS implementation has seen a lot of error, since the start of using it:
How to reproduce
Our NATSService, used by our regular FastAPI app:
And here is is our Dishka setup for the NATS Worker service which will handle NATS messages, the broker is used by the NATSService and the router by the FastAPI FastStream implementation:
And lastly the Worker App:
Expected behavior
We expect our NATSService and FastStream app to have constant connection with the NATS server (hosted on Scaleway), without the errors, and to automatically reconnect which is default behaviour.
Observed behavior
We observe connection errors, so does this suggest that the broker is not reconnecting?:
Environment
Additional context
We have contacted Scaleway support.
They stated that resource limits are:
Which is more than enough, and we are not exceeding this.
They also stated that we must be sure to handle automatic reconnection. But to our understanding this is the default behaviour.
so are we doing something totally wrong with our implementation?
We are not always getting these errors. They seem to appear irregular.
The text was updated successfully, but these errors were encountered: