pingora_cache/
put.rs

1// Copyright 2026 Cloudflare, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Cache Put module
16
17use crate::max_file_size::ERR_RESPONSE_TOO_LARGE;
18use crate::*;
19use bytes::Bytes;
20use http::header;
21use log::warn;
22use pingora_core::protocols::http::{
23    v1::common::header_value_content_length, HttpTask, ServerSession,
24};
25use pingora_error::Error;
26
27/// The interface to define cache put behavior
28pub trait CachePut {
29    /// Return whether to cache the asset according to the given response header.
30    fn cacheable(&self, response: ResponseHeader) -> RespCacheable {
31        let cc = cache_control::CacheControl::from_resp_headers(&response);
32        filters::resp_cacheable(cc.as_ref(), response, false, Self::cache_defaults())
33    }
34
35    /// Return the [CacheMetaDefaults]
36    fn cache_defaults() -> &'static CacheMetaDefaults;
37
38    /// Put interesting things in the span given the parsed response header.
39    fn trace_header(&mut self, _response: &ResponseHeader) {}
40}
41
42use parse_response::ResponseParse;
43
44/// The cache put context
45pub struct CachePutCtx<C: CachePut> {
46    cache_put: C, // the user defined cache put behavior
47    key: CacheKey,
48    storage: &'static (dyn storage::Storage + Sync), // static for now
49    eviction: Option<&'static (dyn eviction::EvictionManager + Sync)>,
50    miss_handler: Option<MissHandler>,
51    max_file_size_tracker: Option<MaxFileSizeTracker>,
52    meta: Option<CacheMeta>,
53    parser: ResponseParse,
54    // FIXME: cache put doesn't have cache lock but some storage cannot handle concurrent put
55    // to the same asset.
56    trace: trace::Span,
57}
58
59impl<C: CachePut> CachePutCtx<C> {
60    /// Create a new [CachePutCtx]
61    pub fn new(
62        cache_put: C,
63        key: CacheKey,
64        storage: &'static (dyn storage::Storage + Sync),
65        eviction: Option<&'static (dyn eviction::EvictionManager + Sync)>,
66        trace: trace::Span,
67    ) -> Self {
68        CachePutCtx {
69            cache_put,
70            key,
71            storage,
72            eviction,
73            miss_handler: None,
74            max_file_size_tracker: None,
75            meta: None,
76            parser: ResponseParse::new(),
77            trace,
78        }
79    }
80
81    /// Set the max cacheable size limit
82    pub fn set_max_file_size_bytes(&mut self, max_file_size_bytes: usize) {
83        self.max_file_size_tracker = Some(MaxFileSizeTracker::new(max_file_size_bytes));
84    }
85
86    async fn put_header(&mut self, meta: CacheMeta) -> Result<()> {
87        #[cfg_attr(not(feature = "trace"), allow(unused_mut))]
88        let mut trace = self.trace.child("cache put header", |o| o.start());
89        let miss_handler = self
90            .storage
91            .get_miss_handler(&self.key, &meta, &trace.handle())
92            .await?;
93        trace::tag_span_with_meta(&mut trace, &meta);
94        self.miss_handler = Some(miss_handler);
95        self.meta = Some(meta);
96        Ok(())
97    }
98
99    async fn put_body(&mut self, data: Bytes, eof: bool) -> Result<()> {
100        // fail if writing the body would exceed the max_file_size_bytes
101        if let Some(size_tracker) = self.max_file_size_tracker.as_mut() {
102            let body_size_allowed = size_tracker.add_body_bytes(data.len());
103            if !body_size_allowed {
104                return Error::e_explain(
105                    ERR_RESPONSE_TOO_LARGE,
106                    format!(
107                        "writing data of size {} bytes would exceed max file size of {} bytes",
108                        data.len(),
109                        size_tracker.max_file_size_bytes(),
110                    ),
111                );
112            }
113        }
114
115        let miss_handler = self.miss_handler.as_mut().unwrap();
116        miss_handler.write_body(data, eof).await
117    }
118
119    async fn finish(&mut self) -> Result<()> {
120        let Some(miss_handler) = self.miss_handler.take() else {
121            // no miss_handler, uncacheable
122            return Ok(());
123        };
124        // Save the entry ID before `finish` consumes the miss handler.
125        let entry_id = miss_handler.entry_id();
126        let finish = miss_handler.finish().await?;
127        if let Some(eviction) = self.eviction.as_ref() {
128            let cache_key = self.key.to_compact();
129            let meta = self.meta.as_ref().unwrap();
130            let entry_key = crate::eviction::CacheEntryKey::from_entry_id(cache_key, entry_id);
131            let evicted = match finish {
132                MissFinishType::Appended(delta, max_size) => {
133                    eviction.increment_weight(&entry_key, delta, max_size)
134                }
135                MissFinishType::Created(size) => {
136                    eviction.admit(entry_key, size, meta.0.internal.fresh_until)
137                }
138            };
139            // actual eviction can be done async
140            let trace = self
141                .trace
142                .child("cache put eviction", |o| o.start())
143                .handle();
144            let storage = self.storage;
145            tokio::task::spawn(async move {
146                for item in evicted {
147                    let target = crate::storage::PurgeTarget::Exact(&item);
148                    if let Err(e) = storage.purge(target, PurgeType::Eviction, &trace).await {
149                        warn!("Failed to purge {target} during eviction for cache put: {e}");
150                    }
151                }
152            });
153        }
154
155        Ok(())
156    }
157
158    fn trace_header(&mut self, header: &ResponseHeader) {
159        self.trace.set_tag(|| {
160            Tag::new(
161                "cache-control",
162                header
163                    .headers
164                    .get_all(http::header::CACHE_CONTROL)
165                    .into_iter()
166                    .map(|v| String::from_utf8_lossy(v.as_bytes()).to_string())
167                    .collect::<Vec<_>>()
168                    .join(","),
169            )
170        });
171    }
172
173    async fn do_cache_put(&mut self, data: &[u8]) -> Result<Option<NoCacheReason>> {
174        let tasks = self.parser.inject_data(data)?;
175        for task in tasks {
176            match task {
177                HttpTask::Header(header, _eos) => {
178                    self.trace_header(&header);
179                    match self.cache_put.cacheable(*header) {
180                        RespCacheable::Cacheable(meta) => {
181                            if let Some(max_file_size_tracker) = &self.max_file_size_tracker {
182                                let content_length_hdr = meta.headers().get(header::CONTENT_LENGTH);
183                                if let Some(content_length) =
184                                    header_value_content_length(content_length_hdr)
185                                {
186                                    if content_length > max_file_size_tracker.max_file_size_bytes()
187                                    {
188                                        return Ok(Some(NoCacheReason::ResponseTooLarge));
189                                    }
190                                }
191                            }
192
193                            self.put_header(meta).await?;
194                        }
195                        RespCacheable::Uncacheable(reason) => {
196                            return Ok(Some(reason));
197                        }
198                    }
199                }
200                HttpTask::Body(data, eos) => {
201                    if let Some(data) = data {
202                        self.put_body(data, eos).await?;
203                    }
204                }
205                _ => {
206                    panic!("unexpected HttpTask during cache put {task:?}");
207                }
208            }
209        }
210        Ok(None)
211    }
212
213    /// Start the cache put logic for the given request
214    ///
215    /// This function will start to read the request body to put into cache.
216    /// Return:
217    /// - `Ok(None)` when the payload will be cache.
218    /// - `Ok(Some(reason))` when the payload is not cacheable
219    pub async fn cache_put(
220        &mut self,
221        session: &mut ServerSession,
222    ) -> Result<Option<NoCacheReason>> {
223        let mut no_cache_reason = None;
224        while let Some(data) = session.read_request_body().await? {
225            if no_cache_reason.is_some() {
226                // even uncacheable, the entire body needs to be drains for 1. downstream
227                // not throwing errors 2. connection reuse
228                continue;
229            }
230            no_cache_reason = self.do_cache_put(&data).await?
231        }
232        self.parser.finish()?;
233        self.finish().await?;
234
235        if let Some(reason) = no_cache_reason {
236            self.trace
237                .set_tag(|| Tag::new("uncacheable_reason", reason.as_str()));
238        }
239
240        Ok(no_cache_reason)
241    }
242}
243
244#[cfg(test)]
245mod test {
246    use super::*;
247    use crate::trace::Span;
248    use once_cell::sync::Lazy;
249
250    struct TestCachePut();
251    impl CachePut for TestCachePut {
252        fn cache_defaults() -> &'static CacheMetaDefaults {
253            const DEFAULT: CacheMetaDefaults =
254                CacheMetaDefaults::new(|_| Some(Duration::from_secs(1)), 1, 1);
255            &DEFAULT
256        }
257    }
258
259    type TestCachePutCtx = CachePutCtx<TestCachePut>;
260    static CACHE_BACKEND: Lazy<MemCache> = Lazy::new(MemCache::new);
261
262    #[tokio::test]
263    async fn test_cache_put() {
264        let key = CacheKey::new("a", "1");
265        let span = Span::inactive();
266        let put = TestCachePut();
267        let mut ctx = TestCachePutCtx::new(put, key.clone(), &*CACHE_BACKEND, None, span);
268        let payload = b"HTTP/1.1 200 OK\r\n\
269        Date: Thu, 26 Apr 2018 05:42:05 GMT\r\n\
270        Content-Type: text/html; charset=utf-8\r\n\
271        Connection: keep-alive\r\n\
272        X-Frame-Options: SAMEORIGIN\r\n\
273        Cache-Control: public, max-age=1\r\n\
274        Server: origin-server\r\n\
275        Content-Length: 4\r\n\r\nrust";
276        // here we skip mocking a real http session for simplicity
277        let res = ctx.do_cache_put(payload).await.unwrap();
278        assert!(res.is_none()); // cacheable
279        ctx.parser.finish().unwrap();
280        ctx.finish().await.unwrap();
281
282        let span = Span::inactive();
283        let (meta, mut hit) = CACHE_BACKEND
284            .lookup(&key, &span.handle())
285            .await
286            .unwrap()
287            .unwrap();
288        assert_eq!(
289            meta.headers().get("date").unwrap(),
290            "Thu, 26 Apr 2018 05:42:05 GMT"
291        );
292        let data = hit.read_body().await.unwrap().unwrap();
293        assert_eq!(data, "rust");
294    }
295
296    #[tokio::test]
297    async fn test_cache_put_uncacheable() {
298        let key = CacheKey::new("a", "1");
299        let span = Span::inactive();
300        let put = TestCachePut();
301        let mut ctx = TestCachePutCtx::new(put, key.clone(), &*CACHE_BACKEND, None, span);
302        let payload = b"HTTP/1.1 200 OK\r\n\
303        Date: Thu, 26 Apr 2018 05:42:05 GMT\r\n\
304        Content-Type: text/html; charset=utf-8\r\n\
305        Connection: keep-alive\r\n\
306        X-Frame-Options: SAMEORIGIN\r\n\
307        Cache-Control: no-store\r\n\
308        Server: origin-server\r\n\
309        Content-Length: 4\r\n\r\nrust";
310        // here we skip mocking a real http session for simplicity
311        let no_cache = ctx.do_cache_put(payload).await.unwrap().unwrap();
312        assert_eq!(no_cache, NoCacheReason::OriginNotCache);
313        ctx.parser.finish().unwrap();
314        ctx.finish().await.unwrap();
315    }
316
317    #[tokio::test]
318    async fn test_cache_put_204_invalid_body() {
319        let key = CacheKey::new("b", "1");
320        let span = Span::inactive();
321        let put = TestCachePut();
322        let mut ctx = TestCachePutCtx::new(put, key.clone(), &*CACHE_BACKEND, None, span);
323        let payload = b"HTTP/1.1 204 OK\r\n\
324        Date: Thu, 26 Apr 2018 05:42:05 GMT\r\n\
325        Content-Type: text/html; charset=utf-8\r\n\
326        Connection: keep-alive\r\n\
327        X-Frame-Options: SAMEORIGIN\r\n\
328        Cache-Control: public, max-age=1\r\n\
329        Server: origin-server\r\n\
330        Content-Length: 4\r\n\r\n";
331        // here we skip mocking a real http session for simplicity
332        let res = ctx.do_cache_put(payload).await.unwrap();
333        assert!(res.is_none()); // cacheable
334                                // 204 should not have body, invalid client input may try to pass one
335        let res = ctx.do_cache_put(b"rust").await.unwrap();
336        assert!(res.is_none()); // still cacheable
337        ctx.parser.finish().unwrap();
338        ctx.finish().await.unwrap();
339
340        let span = Span::inactive();
341        let (meta, mut hit) = CACHE_BACKEND
342            .lookup(&key, &span.handle())
343            .await
344            .unwrap()
345            .unwrap();
346        assert_eq!(
347            meta.headers().get("date").unwrap(),
348            "Thu, 26 Apr 2018 05:42:05 GMT"
349        );
350        // just treated as empty body
351        // (TODO: should we reset content-length/transfer-encoding
352        // headers on 204/304?)
353        let data = hit.read_body().await.unwrap().unwrap();
354        assert!(data.is_empty());
355    }
356
357    #[tokio::test]
358    async fn test_cache_put_extra_body() {
359        let key = CacheKey::new("c", "1");
360        let span = Span::inactive();
361        let put = TestCachePut();
362        let mut ctx = TestCachePutCtx::new(put, key.clone(), &*CACHE_BACKEND, None, span);
363        let payload = b"HTTP/1.1 200 OK\r\n\
364        Date: Thu, 26 Apr 2018 05:42:05 GMT\r\n\
365        Content-Type: text/html; charset=utf-8\r\n\
366        Connection: keep-alive\r\n\
367        X-Frame-Options: SAMEORIGIN\r\n\
368        Cache-Control: public, max-age=1\r\n\
369        Server: origin-server\r\n\
370        Content-Length: 4\r\n\r\n";
371        // here we skip mocking a real http session for simplicity
372        let res = ctx.do_cache_put(payload).await.unwrap();
373        assert!(res.is_none()); // cacheable
374                                // pass in more extra request body that needs to be drained
375        let res = ctx.do_cache_put(b"rustab").await.unwrap();
376        assert!(res.is_none()); // still cacheable
377        let res = ctx.do_cache_put(b"cdef").await.unwrap();
378        assert!(res.is_none()); // still cacheable
379        ctx.parser.finish().unwrap();
380        ctx.finish().await.unwrap();
381
382        let span = Span::inactive();
383        let (meta, mut hit) = CACHE_BACKEND
384            .lookup(&key, &span.handle())
385            .await
386            .unwrap()
387            .unwrap();
388        assert_eq!(
389            meta.headers().get("date").unwrap(),
390            "Thu, 26 Apr 2018 05:42:05 GMT"
391        );
392        let data = hit.read_body().await.unwrap().unwrap();
393        // body only contains specified content-length bounds
394        assert_eq!(data, "rust");
395    }
396}
397
398// maybe this can simplify some logic in pingora::h1
399
400mod parse_response {
401    use super::*;
402    use bstr::ByteSlice;
403    use bytes::BytesMut;
404    use httparse::Status;
405    use pingora_error::{
406        Error,
407        ErrorType::{self, *},
408    };
409
410    pub const INCOMPLETE_BODY: ErrorType = ErrorType::new("IncompleteHttpBody");
411
412    const MAX_HEADERS: usize = 256;
413    const INIT_HEADER_BUF_SIZE: usize = 4096;
414
415    #[derive(Debug, Clone, Copy, PartialEq)]
416    enum ParseState {
417        Init,
418        PartialHeader,
419        PartialBodyContentLength(usize, usize),
420        PartialBody(usize),
421        Done(usize),
422        Invalid(httparse::Error),
423    }
424
425    impl ParseState {
426        fn is_done(&self) -> bool {
427            matches!(self, Self::Done(_))
428        }
429        fn read_header(&self) -> bool {
430            matches!(self, Self::Init | Self::PartialHeader)
431        }
432        fn read_body(&self) -> bool {
433            matches!(
434                self,
435                Self::PartialBodyContentLength(..) | Self::PartialBody(_)
436            )
437        }
438    }
439
440    pub(super) struct ResponseParse {
441        state: ParseState,
442        buf: BytesMut,
443        header_bytes: Bytes,
444    }
445
446    impl ResponseParse {
447        pub fn new() -> Self {
448            ResponseParse {
449                state: ParseState::Init,
450                buf: BytesMut::with_capacity(INIT_HEADER_BUF_SIZE),
451                header_bytes: Bytes::new(),
452            }
453        }
454
455        pub fn inject_data(&mut self, data: &[u8]) -> Result<Vec<HttpTask>> {
456            if self.state.is_done() {
457                // just ignore extra response body after parser is done
458                // could be invalid body appended to a no-content status
459                // or invalid body after content-length
460                // TODO: consider propagating an error to the client
461                return Ok(vec![]);
462            }
463
464            self.put_data(data);
465
466            let mut tasks = vec![];
467            while !self.state.is_done() {
468                if self.state.read_header() {
469                    let header = self.parse_header()?;
470                    let Some(header) = header else {
471                        break;
472                    };
473                    tasks.push(HttpTask::Header(Box::new(header), self.state.is_done()));
474                } else if self.state.read_body() {
475                    let body = self.parse_body()?;
476                    let Some(body) = body else {
477                        break;
478                    };
479                    tasks.push(HttpTask::Body(Some(body), self.state.is_done()));
480                } else {
481                    break;
482                }
483            }
484            Ok(tasks)
485        }
486
487        fn put_data(&mut self, data: &[u8]) {
488            use ParseState::*;
489            if matches!(self.state, Done(_) | Invalid(_)) {
490                panic!("Wrong phase {:?}", self.state);
491            }
492            self.buf.extend_from_slice(data);
493        }
494
495        fn parse_header(&mut self) -> Result<Option<ResponseHeader>> {
496            let mut headers = [httparse::EMPTY_HEADER; MAX_HEADERS];
497            let mut resp = httparse::Response::new(&mut headers);
498            let mut parser = httparse::ParserConfig::default();
499            parser.allow_spaces_after_header_name_in_responses(true);
500            parser.allow_obsolete_multiline_headers_in_responses(true);
501
502            let res = parser.parse_response(&mut resp, &self.buf);
503            let res = match res {
504                Ok(res) => res,
505                Err(e) => {
506                    self.state = ParseState::Invalid(e);
507                    return Error::e_because(
508                        InvalidHTTPHeader,
509                        format!("buf: {:?}", self.buf.as_bstr()),
510                        e,
511                    );
512                }
513            };
514
515            let split_to = match res {
516                Status::Complete(s) => s,
517                Status::Partial => {
518                    self.state = ParseState::PartialHeader;
519                    return Ok(None);
520                }
521            };
522            // safe to unwrap, valid response always has code set.
523            let mut response =
524                ResponseHeader::build(resp.code.unwrap(), Some(resp.headers.len())).unwrap();
525            for header in resp.headers {
526                // TODO: consider hold a Bytes and all header values can be Bytes referencing the
527                // original buffer without reallocation
528                let header_value = pingora_http::header_value_from_slice(header.value);
529                response.append_header(header.name.to_owned(), header_value)?;
530            }
531            // TODO: see above, we can make header value `Bytes` referencing header_bytes
532            let header_bytes = self.buf.split_to(split_to).freeze();
533            self.header_bytes = header_bytes;
534            self.state = body_type(&response);
535
536            Ok(Some(response))
537        }
538
539        fn parse_body(&mut self) -> Result<Option<Bytes>> {
540            use ParseState::*;
541            if self.buf.is_empty() {
542                return Ok(None);
543            }
544            match self.state {
545                Init | PartialHeader | Invalid(_) => {
546                    panic!("Wrong phase {:?}", self.state);
547                }
548                Done(_) => Ok(None),
549                PartialBodyContentLength(total, mut seen) => {
550                    let end = if total < self.buf.len() + seen {
551                        // TODO: warn! more data than expected
552                        total - seen
553                    } else {
554                        self.buf.len()
555                    };
556                    seen += end;
557                    if seen >= total {
558                        self.state = Done(seen);
559                    } else {
560                        self.state = PartialBodyContentLength(total, seen);
561                    }
562                    Ok(Some(self.buf.split_to(end).freeze()))
563                }
564                PartialBody(seen) => {
565                    self.state = PartialBody(seen + self.buf.len());
566                    Ok(Some(self.buf.split().freeze()))
567                }
568            }
569        }
570
571        pub fn finish(&mut self) -> Result<()> {
572            if let ParseState::PartialBody(seen) = self.state {
573                self.state = ParseState::Done(seen);
574            }
575            if !self.state.is_done() {
576                Error::e_explain(INCOMPLETE_BODY, format!("{:?}", self.state))
577            } else {
578                Ok(())
579            }
580        }
581    }
582
583    fn body_type(resp: &ResponseHeader) -> ParseState {
584        use http::StatusCode;
585
586        if matches!(
587            resp.status,
588            StatusCode::NO_CONTENT | StatusCode::NOT_MODIFIED
589        ) {
590            // these status codes cannot have body by definition
591            return ParseState::Done(0);
592        }
593        if let Some(cl) = resp.headers.get(http::header::CONTENT_LENGTH) {
594            // ignore invalid header value
595            if let Some(cl) = std::str::from_utf8(cl.as_bytes())
596                .ok()
597                .and_then(|cl| cl.parse::<usize>().ok())
598            {
599                return if cl == 0 {
600                    ParseState::Done(0)
601                } else {
602                    ParseState::PartialBodyContentLength(cl, 0)
603                };
604            }
605        }
606        // HTTP/1.0 and chunked encoding are both treated as PartialBody
607        // The response body payload should _not_ be chunked encoded
608        // even if the Transfer-Encoding: chunked header is added
609        ParseState::PartialBody(0)
610    }
611
612    #[cfg(test)]
613    mod test {
614        use super::*;
615
616        #[test]
617        fn test_basic_response() {
618            let input = b"HTTP/1.1 200 OK\r\n\r\n";
619            let mut parser = ResponseParse::new();
620            let output = parser.inject_data(input).unwrap();
621            assert_eq!(output.len(), 1);
622            let HttpTask::Header(header, eos) = &output[0] else {
623                panic!("{:?}", output);
624            };
625            assert_eq!(header.status, 200);
626            assert!(!eos);
627
628            let body = b"abc";
629            let output = parser.inject_data(body).unwrap();
630            assert_eq!(output.len(), 1);
631            let HttpTask::Body(data, _eos) = &output[0] else {
632                panic!("{:?}", output);
633            };
634            assert_eq!(data.as_ref().unwrap(), &body[..]);
635            parser.finish().unwrap();
636        }
637
638        #[test]
639        fn test_partial_response_headers() {
640            let input = b"HTTP/1.1 200 OK\r\n";
641            let mut parser = ResponseParse::new();
642            let output = parser.inject_data(input).unwrap();
643            // header is not complete
644            assert_eq!(output.len(), 0);
645
646            let output = parser
647                .inject_data("Server: pingora\r\n\r\n".as_bytes())
648                .unwrap();
649            assert_eq!(output.len(), 1);
650            let HttpTask::Header(header, eos) = &output[0] else {
651                panic!("{:?}", output);
652            };
653            assert_eq!(header.status, 200);
654            assert_eq!(header.headers.get("Server").unwrap(), "pingora");
655            assert!(!eos);
656        }
657
658        #[test]
659        fn test_invalid_headers() {
660            let input = b"HTP/1.1 200 OK\r\nServer: pingora\r\n\r\n";
661            let mut parser = ResponseParse::new();
662            let output = parser.inject_data(input);
663            // header is not complete
664            assert!(output.is_err());
665            match parser.state {
666                ParseState::Invalid(httparse::Error::Version) => {}
667                _ => panic!("should have failed to parse"),
668            }
669        }
670
671        #[test]
672        fn test_body_content_length() {
673            let input = b"HTTP/1.1 200 OK\r\nContent-Length: 6\r\n\r\nabc";
674            let mut parser = ResponseParse::new();
675            let output = parser.inject_data(input).unwrap();
676
677            assert_eq!(output.len(), 2);
678            let HttpTask::Header(header, _eos) = &output[0] else {
679                panic!("{:?}", output);
680            };
681            assert_eq!(header.status, 200);
682
683            let HttpTask::Body(data, eos) = &output[1] else {
684                panic!("{:?}", output);
685            };
686            assert_eq!(data.as_ref().unwrap(), "abc");
687            assert!(!eos);
688
689            let output = parser.inject_data(b"def").unwrap();
690            assert_eq!(output.len(), 1);
691            let HttpTask::Body(data, eos) = &output[0] else {
692                panic!("{:?}", output);
693            };
694            assert_eq!(data.as_ref().unwrap(), "def");
695            assert!(eos);
696
697            parser.finish().unwrap();
698        }
699
700        #[test]
701        fn test_body_chunked() {
702            let input = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nrust";
703            let mut parser = ResponseParse::new();
704            let output = parser.inject_data(input).unwrap();
705
706            assert_eq!(output.len(), 2);
707            let HttpTask::Header(header, _eos) = &output[0] else {
708                panic!("{:?}", output);
709            };
710            assert_eq!(header.status, 200);
711
712            let HttpTask::Body(data, eos) = &output[1] else {
713                panic!("{:?}", output);
714            };
715            assert_eq!(data.as_ref().unwrap(), "rust");
716            assert!(!eos);
717
718            parser.finish().unwrap();
719        }
720
721        #[test]
722        fn test_body_content_length_early() {
723            let input = b"HTTP/1.1 200 OK\r\nContent-Length: 6\r\n\r\nabc";
724            let mut parser = ResponseParse::new();
725            let output = parser.inject_data(input).unwrap();
726
727            assert_eq!(output.len(), 2);
728            let HttpTask::Header(header, _eos) = &output[0] else {
729                panic!("{:?}", output);
730            };
731            assert_eq!(header.status, 200);
732
733            let HttpTask::Body(data, eos) = &output[1] else {
734                panic!("{:?}", output);
735            };
736            assert_eq!(data.as_ref().unwrap(), "abc");
737            assert!(!eos);
738
739            parser.finish().unwrap_err();
740        }
741
742        #[test]
743        fn test_body_content_length_more_data() {
744            let input = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nabc";
745            let mut parser = ResponseParse::new();
746            let output = parser.inject_data(input).unwrap();
747
748            assert_eq!(output.len(), 2);
749            let HttpTask::Header(header, _eos) = &output[0] else {
750                panic!("{:?}", output);
751            };
752            assert_eq!(header.status, 200);
753
754            let HttpTask::Body(data, eos) = &output[1] else {
755                panic!("{:?}", output);
756            };
757            assert_eq!(data.as_ref().unwrap(), "ab");
758            assert!(eos);
759
760            // extra data is dropped without error
761            parser.finish().unwrap();
762        }
763
764        #[test]
765        fn test_body_chunked_partial_chunk() {
766            let input = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nru";
767            let mut parser = ResponseParse::new();
768            let output = parser.inject_data(input).unwrap();
769
770            assert_eq!(output.len(), 2);
771            let HttpTask::Header(header, _eos) = &output[0] else {
772                panic!("{:?}", output);
773            };
774            assert_eq!(header.status, 200);
775
776            let HttpTask::Body(data, eos) = &output[1] else {
777                panic!("{:?}", output);
778            };
779            assert_eq!(data.as_ref().unwrap(), "ru");
780            assert!(!eos);
781
782            let output = parser.inject_data(b"st\r\n").unwrap();
783            assert_eq!(output.len(), 1);
784            let HttpTask::Body(data, eos) = &output[0] else {
785                panic!("{:?}", output);
786            };
787            assert_eq!(data.as_ref().unwrap(), "st\r\n");
788            assert!(!eos);
789        }
790
791        #[test]
792        fn test_no_body_content_length() {
793            let input = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n";
794            let mut parser = ResponseParse::new();
795            let output = parser.inject_data(input).unwrap();
796
797            assert_eq!(output.len(), 1);
798            let HttpTask::Header(header, eos) = &output[0] else {
799                panic!("{:?}", output);
800            };
801            assert_eq!(header.status, 200);
802            assert!(eos);
803
804            parser.finish().unwrap();
805        }
806
807        #[test]
808        fn test_no_body_304_no_content_length() {
809            let input = b"HTTP/1.1 304 Not Modified\r\nCache-Control: public, max-age=10\r\n\r\n";
810            let mut parser = ResponseParse::new();
811            let output = parser.inject_data(input).unwrap();
812
813            assert_eq!(output.len(), 1);
814            let HttpTask::Header(header, eos) = &output[0] else {
815                panic!("{:?}", output);
816            };
817            assert_eq!(header.status, 304);
818            assert!(eos);
819
820            parser.finish().unwrap();
821        }
822
823        #[test]
824        fn test_204_with_chunked_body() {
825            let input = b"HTTP/1.1 204 No Content\r\nCache-Control: public, max-age=10\r\nTransfer-Encoding: chunked\r\n\r\n";
826            let mut parser = ResponseParse::new();
827            let output = parser.inject_data(input).unwrap();
828
829            assert_eq!(output.len(), 1);
830            let HttpTask::Header(header, eos) = &output[0] else {
831                panic!("{:?}", output);
832            };
833            assert_eq!(header.status, 204);
834            assert!(eos);
835
836            // 204 should not have a body, parser ignores bad input
837            let output = parser.inject_data(b"4\r\nrust\r\n0\r\n\r\n").unwrap();
838            assert!(output.is_empty());
839            parser.finish().unwrap();
840        }
841
842        #[test]
843        fn test_204_with_content_length() {
844            let input = b"HTTP/1.1 204 No Content\r\nCache-Control: public, max-age=10\r\nContent-Length: 4\r\n\r\n";
845            let mut parser = ResponseParse::new();
846            let output = parser.inject_data(input).unwrap();
847
848            assert_eq!(output.len(), 1);
849            let HttpTask::Header(header, eos) = &output[0] else {
850                panic!("{:?}", output);
851            };
852            assert_eq!(header.status, 204);
853            assert!(eos);
854
855            // 204 should not have a body, parser ignores bad input
856            let output = parser.inject_data(b"rust").unwrap();
857            assert!(output.is_empty());
858            parser.finish().unwrap();
859        }
860
861        #[test]
862        fn test_200_with_zero_content_length_more_data() {
863            let input = b"HTTP/1.1 200 OK\r\nCache-Control: public, max-age=10\r\nContent-Length: 0\r\n\r\n";
864            let mut parser = ResponseParse::new();
865            let output = parser.inject_data(input).unwrap();
866
867            assert_eq!(output.len(), 1);
868            let HttpTask::Header(header, eos) = &output[0] else {
869                panic!("{:?}", output);
870            };
871            assert_eq!(header.status, 200);
872            assert!(eos);
873
874            let output = parser.inject_data(b"rust").unwrap();
875            assert!(output.is_empty());
876            parser.finish().unwrap();
877        }
878    }
879}
pingora_cache/put.rs

pingora_cache/
put.rs
