stream-transfer-limit 0.1.0

Byte-count transfer limits for fallible futures streams
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

# stream-transfer-limit

`stream-transfer-limit` applies cumulative byte limits to fallible
`futures::Stream`s.

It is meant for cases where an HTTP middleware or fixed body limit is not quite
the right abstraction: for example, when the allowed transfer size is chosen
after routing, authentication, configuration, or account lookup.

```rust
use futures::{stream, StreamExt};
use stream_transfer_limit::{TransferLimit, TransferLimitError};

fn main() {
    futures::executor::block_on(async {
        let chunks = stream::iter([
            Ok::<_, std::io::Error>(vec![1, 2]),
            Ok(vec![3, 4, 5]),
        ]);

        let max_bytes = 4;
        let mut limited = TransferLimit::new(max_bytes).wrap(chunks);

        assert_eq!(limited.next().await.unwrap().unwrap(), vec![1, 2]);
        assert!(matches!(
            limited.next().await.unwrap(),
            Err(TransferLimitError::LimitExceeded { limit: 4, actual: 5 })
        ));
        assert!(limited.next().await.is_none());
    });
}
```

## Behavior

- The stream may produce exactly the configured limit.
- The first chunk that makes the cumulative byte count greater than the limit
  returns `TransferLimitError::LimitExceeded`.
- After a limit error, the wrapper terminates and does not keep polling the
  inner stream.
- The default byte counter is `usize`. If the cumulative count cannot fit in the
  selected counter type, the wrapper returns `TransferLimitError::CounterOverflow`
  and terminates.
- Inner stream errors are preserved as `TransferLimitError::Inner`.
- Progress callbacks receive cumulative bytes after every successful chunk read
  from the inner stream, including the chunk that crosses the limit.

## APIs

For new code, construct a `TransferLimit` and wrap the stream:

```rust
use futures::stream;
use stream_transfer_limit::TransferLimit;

fn main() {
    let chunks = stream::iter([Ok::<_, std::io::Error>(vec![0; 512])]);

    let _limited = TransferLimit::new(1024)
        .on_progress(|bytes_seen| {
            eprintln!("read {bytes_seen} bytes");
        })
        .wrap(chunks);
}
```

`usize` avoids unnecessary conversions for ordinary in-memory or platform-sized
limits. For very large streams, choose a wider counter explicitly:

```rust
use futures::stream;
use stream_transfer_limit::TransferLimit;

fn main() {
    let chunks = stream::iter([Ok::<_, std::io::Error>(vec![0])]);
    let huge_limit = 16_u128 * 1024 * 1024 * 1024 * 1024;

    let _limited = TransferLimit::<u128>::from_limit(huge_limit).wrap(chunks);
}
```

The crate implements counters for `usize`, `u64`, and `u128`.

## Development

The repository includes a Nix flake for a reproducible development shell:

```sh
nix develop
```

If you use direnv, allow the checked-in `.envrc` once:

```sh
direnv allow
```

The development shell provides the pinned Rust toolchain, rustfmt, clippy,
rust-analyzer, cargo helper tools, Docker client, and `act`.

Run the local CI pipeline:

```sh
scripts/ci.sh
```

Run the GitHub Actions workflow locally with `act`:

```sh
scripts/act-ci.sh
```

## Why not tower-http?

`tower_http::limit::RequestBodyLimitLayer` and `http_body_util::Limited` are
good choices when the limit is known at the HTTP-body layer. This crate is for
code that already has a `futures::Stream` and needs to choose the limit closer to
business logic, such as per-tenant transfer limits.