notedthat-write 0.12.1

Shared write path (commit, patch, replace) for NotedThat HTTP API and WebDAV surfaces
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

use bytes::Bytes;
use notedthat_core::StorageError;
use notedthat_indexer::IndexEvent;
use tokio::sync::mpsc;

use super::support::{
    ReplaceArgs, Script, TestStorage, conditionals, expect_replace_err, kb, path, run_replace,
    run_replace_with,
};

#[tokio::test]
async fn stale_caller_if_match_returns_precondition_failed_no_retry() {
    let storage = TestStorage::with_body(b"hello world");
    let err = expect_replace_err(
        run_replace_with(
            &storage,
            ReplaceArgs::one("world", "planet"),
            conditionals(Some("\"stale\"")),
            1024,
            None,
        )
        .await,
    );

    assert!(matches!(
        err,
        crate::WriteError::Storage(StorageError::PreconditionFailed)
    ));
    let calls = storage.calls();
    assert_eq!(calls.head, 1);
    assert_eq!(calls.get, 0);
    assert_eq!(calls.put, 0);
    assert_eq!(storage.body(), Bytes::from_static(b"hello world"));
}

#[tokio::test]
async fn window_412_from_get_absorbed_by_two_retries_then_surfaces_precondition_failed() {
    for failures in [1, 2] {
        let storage = TestStorage::with_script(
            b"hello world",
            Script {
                get_failures_remaining: failures,
                ..Script::default()
            },
        );

        run_replace(&storage, ReplaceArgs::one("world", "planet"))
            .await
            .expect("GET precondition window is absorbed before max attempts");

        assert_eq!(storage.body(), Bytes::from_static(b"hello planet"));
    }

    let storage = TestStorage::with_script(
        b"hello world",
        Script {
            get_failures_remaining: 3,
            ..Script::default()
        },
    );
    let err = expect_replace_err(run_replace(&storage, ReplaceArgs::one("world", "planet")).await);

    assert!(matches!(
        err,
        crate::WriteError::Storage(StorageError::PreconditionFailed)
    ));
    let calls = storage.calls();
    assert_eq!(calls.head, 3);
    assert_eq!(calls.get, 3);
    assert_eq!(calls.put, 0);
}

#[tokio::test]
async fn window_412_from_put_absorbed_by_two_retries() {
    let storage = TestStorage::with_script(
        b"hello world",
        Script {
            put_failures_remaining: 2,
            ..Script::default()
        },
    );

    run_replace(&storage, ReplaceArgs::one("world", "planet"))
        .await
        .expect("PUT precondition window is absorbed before max attempts");

    assert_eq!(storage.body(), Bytes::from_static(b"hello planet"));
    let calls = storage.calls();
    assert_eq!(calls.head, 3);
    assert_eq!(calls.get, 3);
    assert_eq!(calls.put, 3);
}

#[tokio::test]
async fn indexer_channel_full_returns_backpressure_upsert_after_successful_put() {
    let storage = TestStorage::with_body(b"hello world");
    let (indexer_tx, _rx) = mpsc::channel(1);
    indexer_tx
        .try_send(IndexEvent::Upsert {
            kb: kb(),
            object_key: path(),
            etag: "queued".to_string(),
            mtime: 0,
        })
        .expect("prefill indexer channel");
    let err = expect_replace_err(
        run_replace_with(
            &storage,
            ReplaceArgs::one("world", "planet"),
            conditionals(Some("etag1")),
            1024,
            Some(indexer_tx),
        )
        .await,
    );

    assert!(matches!(err, crate::WriteError::IndexerBackpressureUpsert));
    assert_eq!(
        storage.read().await.bytes,
        Bytes::from_static(b"hello planet")
    );
}