Strange behaviour when stream PublisherID is too long

[deadtrickster]

Describe the bug

Two bugs in one:
First:
PublisherID length not checked by stream plugin front end. It happily serializes it and PublisherID ends up in a osiris_tracking:add call which checks theTrkId lenght. And then crashes with function_clause.

Also I get no slap from python client. It happily keeps running connection. While when say rabbit is dead or gen_batch_server is busy from breakpoints I get connection timeout

Second (probably to be moved to osiris directly):

I also got this when sending a smaller PublisherId of 256 bytes:

2024-10-09 13:29:44.217235+02:00 [debug] <0.1181.0> rabbit_stream_coordinator: running action: 'stop' for __hello-python-stream-small3_1728473315576504036 on node rabbit@arch in epoch 1874
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>     supervisor: {local,osiris_server_sup}
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>     errorContext: child_terminated
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>     reason: {function_clause,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                 [{osiris_tracking,parse_trailer,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                      [<<87,68,109,107,122,90,97,82,75,85,73,113,122,69,79,120,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         117,120,109,101,65,121,83,102,85,108,102,120,107,71,71,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         89,108,97,101,70,106,115,98,109,69,103,100,107,118,115,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         86,72,99,67,122,99,98,71,114,100,70,102,71,75,65,70,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         122,116,75,83,90,97,120,121,109,79,105,121,102,72,69,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         107,68,77,68,76,103,88,78,119,110,66,102,100,75,120,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         116,111,114,83,113,108,83,106,78,106,114,80,89,74,79,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         71,74,80,70,76,81,78,87,109,76,71,82,112,77,119,69,98,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         120,66,106,78,88,76,76,110,101,67,90,102,122,114,107,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         88,87,116,71,80,98,105,97,115,103,104,114,111,121,107,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         67,73,101,106,110,99,101,74,108,83,122,90,76,119,102,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         77,113,103,86,112,118,114,119,109,101,104,89,115,86,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         110,79,112,71,116,109,76,77,68,118,106,68,84,73,79,122,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         119,108,90,82,114,111,69,98,110,106,69,110,85,104,100,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         109,85,118,97,103,97,113,118,85,66,86,114,80,111,98,68,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         109,111,120,76,72,70,101,102,88,111,117,66,70,90,105,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         87,121,48,0,0,0,0,0,0,0,1>>,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                       7,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                       {osiris_tracking,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           {cfg,255},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           #{sequences => #{},offsets => #{},timestamps => #{}},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           #{<<>> => {7,6147522654068625751},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe0">> => {0,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe1">> => {1,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe2">> => {2,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe3">> => {3,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe4">> => {4,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe5">> => {5,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe6">> => {6,1}},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           #{},#{}}],

Reproduction steps

iimport asyncio
import random
import string
from rstream import Producer

STREAM_NAME = "hello-python-stream-small3"
# 5GB
STREAM_RETENTION = 500

message =''.join(random.choices(string.ascii_letters, k=4))
message_bytes = message.encode()


async def send():
    
    i = 0
    while True:
        async with Producer(
                host="localhost",
                username="guest",
                password="guest",
        ) as producer:
            await producer.create_stream(
                STREAM_NAME, exists_ok=True, arguments={"max-length-bytes": STREAM_RETENTION}
            )

            #await producer.send(stream=STREAM_NAME, publisher_name="qwe{}".format(i), message=message_bytes)

            #await producer.send(stream=STREAM_NAME, publisher_name=''.join(random.choices(string.ascii_letters, k=4096)), message=message_bytes)
            #await producer.send(stream=STREAM_NAME, publisher_name="tvyOZOJWJEZdOPQRGLgJXbUQIxHDlMdaLIdpRCYViiNYvpTkyvNeXHHeDBHjVLzdEJoJNRyRVvFoDPaZQFBVyEwGkMvLbtppYkGRSvPJBClcFxdZrHSmpDNiZYdbiITtxoalsEPwYVIKYBVHnqnoUxRBUEbBXKhLHaEKVvJPdeunrcHuWwrFbdBxNEszlbMMflqLfUknSpWFanftkgDHvffuEajEnqqIZbtQAZXtqmEysubzHPILuubQaupmTykT", message=message_bytes)
            
            print(" [x] Hello, World! message {} sent".format(i))
            i += 1
            
            # input(" [x] Press Enter to close the producer  ...")
            await producer.close()

with asyncio.Runner() as runner:
    runner.run(send())

uncomment the third send (length of 256 and constant) and then second to get writer restarted. restarting Rabbit also gives same result

Expected behavior

Stream writer doesn't crash, Python client yells at me probably with proper errorcode

Additional context

2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>   crasher:
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     initial call: osiris_writer:init/1
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     pid: <0.902.0>
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     registered_name: []
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     exception exit: {{case_clause,
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>                       {'EXIT',
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>                        {function_clause,
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>                         [{osiris_tracking,add,