vein 0.4.0

A fast, intelligent RubyGems proxy/mirror server written in Rust
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
123
124
125
126
127
128
129
130
131

use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use rama::{
    Layer, Service,
    error::OpaqueError,
    http::{
        BodyExtractExt, HeaderValue, Request, Response, StatusCode, client::EasyHttpWebClient,
        header, layer::required_header::AddRequiredRequestHeadersLayer,
        service::client::HttpClientExt as _,
    },
    layer::{MapErrLayer, TimeoutLayer},
    telemetry::tracing::{error, info},
};
use vein_adapter::CacheBackendKind;

const NAMES_URL: &str = "https://rubygems.org/names.gz";
const META_ETAG: &str = "catalog_names_etag";
const META_LAST_MODIFIED: &str = "catalog_names_last_modified";

pub async fn sync_names_once(index: &CacheBackendKind) -> Result<Option<usize>> {
    let client = build_client()?;
    sync_names_with_client(index, &client).await
}

fn build_client() -> Result<impl Service<Request, Output = Response, Error = OpaqueError>> {
    // NOTE if you want pooling you'll have to
    // use build_connector to also include the pool desired
    let inner = EasyHttpWebClient::default();

    // decompression support would be added as layer on top

    Ok((
        MapErrLayer::new(OpaqueError::from_boxed),
        TimeoutLayer::new(Duration::from_secs(60)),
        AddRequiredRequestHeadersLayer::new()
            .with_user_agent_header_value(HeaderValue::from_static("vein-catalog/0.1.0")),
    )
        .into_layer(inner))
}

async fn sync_names_with_client(
    index: &CacheBackendKind,
    client: &impl Service<Request, Output = Response, Error = OpaqueError>,
) -> Result<Option<usize>> {
    let mut request = client.get(NAMES_URL);

    if let Some(etag) = index.catalog_meta_get(META_ETAG).await? {
        request = request.header(header::IF_NONE_MATCH, etag);
    }

    if let Some(last_modified) = index.catalog_meta_get(META_LAST_MODIFIED).await? {
        request = request.header(header::IF_MODIFIED_SINCE, last_modified);
    }

    let response = request
        .send()
        .await
        .context("requesting rubygems names list")?;

    if response.status() == StatusCode::NOT_MODIFIED {
        info!("catalog names list is up to date");
        return Ok(None);
    }

    if !response.status().is_success() {
        return Err(anyhow::anyhow!("fetching rubygems names list"));
    };

    let etag = response
        .headers()
        .get(header::ETAG)
        .and_then(|value| value.to_str().ok())
        .map(ToOwned::to_owned);
    let last_modified = response
        .headers()
        .get(header::LAST_MODIFIED)
        .and_then(|value| value.to_str().ok())
        .map(ToOwned::to_owned);

    let text = response
        .try_into_string()
        .await
        .context("decoding names list")?;

    let names: Vec<String> = text
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .map(|line| line.to_owned())
        .collect();

    let total = names.len();
    for chunk in names.chunks(1_000) {
        index.catalog_upsert_names(chunk).await?;
    }

    if let Some(etag) = etag {
        index.catalog_meta_set(META_ETAG, &etag).await?;
    }
    if let Some(last_modified) = last_modified {
        index
            .catalog_meta_set(META_LAST_MODIFIED, &last_modified)
            .await?;
    }

    info!(total, "catalog names synced");

    Ok(Some(total))
}

/// Spawns a background task that periodically syncs the catalog names.
/// Syncs every 6 hours.
pub fn spawn_background_sync(backend: Arc<CacheBackendKind>) -> Result<()> {
    tokio::spawn(async move {
        let sync_interval = Duration::from_secs(6 * 60 * 60); // 6 hours

        loop {
            match sync_names_once(backend.as_ref()).await {
                Ok(Some(count)) => info!(count, "background catalog sync completed"),
                Ok(None) => info!("background catalog sync: already up to date"),
                Err(e) => error!(error = %e, "background catalog sync failed"),
            }

            tokio::time::sleep(sync_interval).await;
        }
    });

    Ok(())
}