Delete old objects when calling ObjectStore::put

[thinety]

When an object already exists in the object store, ObjectStore::put is supposed to purge it. Currently, this does not happen because the object name is erroneously double-encoded.

To reproduce:

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    let client = async_nats::connect("localhost:4222").await?;
    let jetstream = async_nats::jetstream::new(client);
    let kv = jetstream
        .create_object_store(async_nats::jetstream::object_store::Config {
            bucket: "store".to_owned(),
            ..Default::default()
        })
        .await?;

    let mut file = tokio::fs::File::open("foo").await?;
    kv.put("file", &mut file).await?;

    let mut file = tokio::fs::File::open("bar").await?;
    kv.put("file", &mut file).await?;

    Ok(())
}

echo foo >foo
echo bar >bar
cargo run
nats stream view OBJ_store

Output:

[1] Subject: $O.store.C.pGkr4eEuZBEg3NrGeApigQ Received: 2025-01-27T20:43:37-03:00
foo

[3] Subject: $O.store.C.pGkr4eEuZBEg3NrGeApioo Received: 2025-01-27T20:43:37-03:00
bar

[4] Subject: $O.store.M.ZmlsZQ== Received: 2025-01-27T20:43:37-03:00

  Nats-Rollup: sub
{"name":"file","description":null,"metadata":{},"headers":null,"options":{"link":null,"max_chunk_size":131072},"bucket":"store","nuid":"pGkr4eEuZBEg3NrGeApioo","size":4,"chunks":1,"mtime":"2025-01-27T23:43:37.168071192Z","digest":"SHA-256=fYZelZskZpGMmGOvypQtD7idfJrAyZuvw3SVBN7ZdzA="}


20:43:39 Reached apparent end of data

After the changes introduced by this PR:

[3] Subject: $O.store.C.LrStiyrrO355ash9mfODWe Received: 2025-01-27T20:44:08-03:00
bar

[4] Subject: $O.store.M.ZmlsZQ== Received: 2025-01-27T20:44:08-03:00

  Nats-Rollup: sub
{"name":"file","description":null,"metadata":{},"headers":null,"options":{"link":null,"max_chunk_size":131072},"bucket":"store","nuid":"LrStiyrrO355ash9mfODWe","size":4,"chunks":1,"mtime":"2025-01-27T23:44:08.885415879Z","digest":"SHA-256=fYZelZskZpGMmGOvypQtD7idfJrAyZuvw3SVBN7ZdzA="}


20:44:12 Reached apparent end of data

[Jarema]

Thanks for catching and fixing this!

Would be really nice to have a test for it that checks the stream content after delete.
Are you able to contribute one?

[thinety]

Added a test object_store::purge_existing_object_on_put that checks for the purge of the old object.

I'm not entirely sure how to do it without knowledge of the underlying implementation (stream name, chunk and meta topics). Let me know if you have a better idea.

[Jarema]

That is good enough.

LGTM!

[thinety]

I'm sorry, I just realized that I shouldn't have moved the test for validity of the (encoded) object name, which is purely client-side logic, to the middle of the function.

My intention was to put the definition close to the use of the variable, but in this case I believe that it's wrong to return such an error after interaction with the server.

Should I open another PR?