fix/topic streaming bug

scramjetorg / transform-hub

Scramjet Transform Hub (STH) is a runtime supervisor that can run data processing programs called Sequences and manage local resources on any Linux server, Docker on small edge servers, and even large-scale Kubernetes clusters in the cloud or datacenters. It connects to Scramjet Spaces in Scramjet Cloud Platform.

GNU Affero General Public License v3.0

67 stars 8 forks source link

fix/topic streaming bug #932

Closed MichalCz closed 1 year ago

MichalCz commented 1 year ago

What?

Fix CLI not to read space state when it's not needed
Fix API server disconnection tracking

Why?

Topics were wrongly redirected to new instances on every request. This fixes the issue.

Also makes topic writers queue up after each other - first topic writer needs to be consumed, before others can start writing.

Also audit stream logs were turned off - these cause a loop of messages.

Usage:

In separate terminals:

yarn start:dev
cat /dev/random | hexdump | head -n 10 | si topic send mytopic
cat /dev/random | hexdump | head -n 10 | si topic send mytopic # yes, again
# wait a little - above terminals should wait and topics should wait
si topic get mytopic
# topics should be consumed only after this command

Clickup Task:

How it works:

see consumePipe in sd-adapter.
see changes in api-server (pause)

Review checks:

These aspects need to be checked by the reviewer:

[ ] Verify and confirm operation (please post a screenshot)
[ ] All STH tests pass
[ ] All Scramjet Cloud Platform tests pass
[x] Documentation is updated or no changes

a-tylenda commented 1 year ago

This PR additionally fixes two issues:

but:

There is no Error returned (415) when the topic is sent with the application/x-ndjson content type and collected with the text/plain content type:

In GHA we can notice that sometimes error 415 is returned and sometimes not:

passed test in GHA → https://github.com/scramjetorg/transform-hub/actions/runs/6001225734/job/16276363240#step:10:41
failed test in GHA → https://github.com/scramjetorg/transform-hub/actions/runs/6001208318/job/16275059228#step:10:44

both jobs were done on the same commit of course.

What is more, this error is also returned when you try another way around, meaning:

First si topic get pets (default content type text/plain is used)
Second si topic send pets -t application/x-ndjson (overwritten content type)

a-tylenda commented 1 year ago

Also, the method getNamedData() has changed so that now, when you use it on hostClient you need to pass more arguments than just the topic name as it used to be before.

hostClient.getNamedData(topic, {}, "text/plain")

otherwise, you will get a bad request error 400: