xensec_internal_tools/
cache.rs

1// Copyright 2021 Citrix
2// SPDX-License-Identifier: MIT OR Apache-2.0
3// There is NO WARRANTY.
4
5use crate::prelude::*;
6
7use rusqlite::{/*params,*/ Connection, Transaction, ToSql, types::ValueRef};
8
9type RE = rusqlite::Error;
10
11const MAX_RETRIES: u32 = 10;
12
13#[derive(Debug,Copy,Clone)]
14#[derive(Hash,Eq,PartialEq,Ord,PartialOrd)]
15#[derive(EnumIter)]
16pub enum RefreshKind {
17  XsaGit,
18  XsaParsed,
19  CveListXenSec,
20  CveListUpstream,
21  UpstreamMr(XsaNum),
22}
23
24type Refreshed = BTreeSet<Refresh>;
25
26#[derive(Debug,Copy,Clone)]
27#[derive(Hash,Eq,PartialEq,Ord,PartialOrd)]
28pub struct Refresh(RefreshKind /* XsaNum is actually set */);
29
30impl RefreshKind {
31  pub fn bind(self, xsa: Option<XsaNum>) -> Refresh {
32    use RefreshKind as R;
33    Refresh(match self {
34      R::UpstreamMr(old) => { assert_eq!(old,0); R::UpstreamMr(xsa.unwrap()) },
35      other => other,
36    })
37  }
38}
39
40#[derive(Debug,Copy,Clone)]
41#[derive(Hash,Eq,PartialEq,Ord,PartialOrd)]
42pub struct RefreshingToken(Refresh);
43
44impl Config {
45  pub fn want_refresh(&self, rk: RefreshKind, xsa: Option<XsaNum>)
46                      -> Option<RefreshingToken>
47  {
48    use RefreshKind as R;
49    let want = match rk {
50      R::XsaGit |
51      R::CveListXenSec |
52      R::CveListUpstream => ! self.co.no_pull,
53      R::XsaParsed |
54      R::UpstreamMr(_) => self.co.refresh,
55    };
56    if want {
57      Some(RefreshingToken(rk.bind(xsa)))
58    } else {
59      None
60    }
61  }
62}
63
64impl Ctx {
65  #[throws(AE)]
66  pub fn want_remote_actions(&self, xsa: XsaNum,
67                             desc: &str, info: &dyn Debug)
68                             -> bool
69  {
70    use RefreshKind as R;
71    let want = ! self.cfg.co.dry_run;
72    if want {
73      if self.cfg.co.no_pull { throw!(anyhow!(
74          "--no-pull specified with action which makes remote updats"
75      )); }
76
77      for rf in RefreshKind::iter() {
78        let rf = rf.bind(Some(xsa));
79        match rf {
80          Refresh(R::XsaParsed) => {
81            // invalidation is automatic based on version
82          },
83          rf => {
84            assert!(self.cache.refreshed.contains(&rf),
85                    "program logic error, did not refresh {:?}, did {:?}",
86                    rf, &self.cache.refreshed);
87          }
88        }
89      }
90    } else {
91      eprintln!("dry run, would {} {:?}", desc, info);
92    }
93    want
94  }
95} 
96
97impl Cache {
98  pub fn did_refresh(&mut self, rt: Option<RefreshingToken>) {
99    if let Some(rt) = rt {
100      self.refreshed.insert(rt.0);
101    }
102  }
103}
104
105static SCHEMA: &[&str] = &[r#"
106        CREATE TABLE IF NOT EXISTS xsagit_parsed_1 (
107            xsa  INTEGER PRIMARY KEY,
108            vsn  INTEGER NOT NULL,
109            json TEXT    NOT NULL
110        );
111"#,r#"
112        CREATE TABLE IF NOT EXISTS cvelist_mr_1 (
113            xsa  INTEGER PRIMARY KEY,
114            json TEXT    NOT NULL
115        );            
116"#];
117
118pub trait Cacheable: Serialize + DeserializeOwned + Debug {
119  type Key: Debug;
120  const TABLE: &'static str;
121  const KEY_KCOLS: &'static [&'static str];
122  const KEY_VCOLS: &'static [&'static str];
123  const REFRESH: RefreshKind;
124  fn key_col_values(key: &Self::Key) -> Result<Vec<&dyn ToSql>, AE>;
125}
126
127impl Cacheable for xsagit::Parsed {
128  // Invalidation:
129  //
130  // Cache will automatically be invalidated by new xsa publication,
131  // because Vsn is in KEY_VCOLS.
132  //
133  // If the xsa.git code changes in ways which change the generated
134  // Parsed, this won't be detected and a manual invalidation will be
135  // needed.
136  type Key = (XsaNum, XsaVsn);
137  const TABLE: &'static str = "xsagit_parsed_1";
138  const KEY_KCOLS: &'static [&'static str] = &["xsa"];
139  const KEY_VCOLS: &'static [&'static str] = &["vsn"];
140  const REFRESH: RefreshKind = RefreshKind::XsaParsed;
141  #[throws(AE)]
142  fn key_col_values<'k>((xsa, vsn): &'k (XsaNum, XsaVsn))  -> Vec<&'k dyn ToSql>
143  { vec![xsa as _, vsn as _] }
144}
145
146impl Cacheable for upstream::MrState {
147  // Invalidation:
148  //
149  // Merge requets are supposed to be created only by us.  If they
150  // are created manually, the cache must be manually invalidated.
151  //
152  // When (before) we create an MR, code here invalidates the cache
153  // and recheck the status.
154  //
155  // If an MR is merged this will show up as the CVEs being Populated
156  // which likewise means we can invalidate the cache.
157  //
158  // If an MR is closed, we don't detect this.  We will have a cron
159  // job to refresh the status.
160  type Key = XsaNum;
161  const TABLE: &'static str = "cvelist_mr_1";
162  const KEY_KCOLS: &'static [&'static str] = &["xsa"];
163  const KEY_VCOLS: &'static [&'static str] = &[];
164  const REFRESH: RefreshKind = RefreshKind::UpstreamMr(0);
165  #[throws(AE)]
166  fn key_col_values<'k>(xsa: &'k XsaNum)  -> Vec<&'k dyn ToSql>
167  { vec![xsa as _] }
168}
169
170#[derive(Debug)]
171pub struct Cache {
172  pub cfg: Arc<Config>,
173  pub refreshed: Refreshed,
174  db: Connection,
175}
176
177#[derive(Debug,Copy,Clone)]
178pub struct XsaDoneData {
179  done: Option<XsaVsn>,
180}
181
182#[throws(fmt::Error)]
183fn write_cond_where_cols<'k,S,C>(mut stmt: S, cols: C)
184where S: fmt::Write,
185      C: Iterator<Item=&'k str>
186{
187  for (k, conj) in izip!(
188    cols,
189    iter::once("WHERE").chain(iter::repeat("AND")),
190  ) {
191    write!(stmt, " {} {}=?", conj, k)?;
192  }
193}
194
195impl Cache {
196  #[throws(AE)]
197  pub fn new(cfg: &Arc<Config>) -> Self {
198    let db = rusqlite::Connection::open(&cfg.done_cache)
199      .with_context(|| cfg.done_cache.clone()).context("opne db")?;
200
201    for stmt in SCHEMA {
202      db.execute(stmt,[])
203        .with_context(|| stmt.to_owned())
204        .context("ensure schema is set up")?;
205    }
206
207    Cache { db, cfg: cfg.clone(), refreshed: default() }
208  }
209
210  #[throws(AE)]
211  pub fn with<F,O>(&mut self, mut f: F) -> O
212  where F: FnMut(&mut Transaction) -> Result<O, AE>
213  {
214    use rusqlite::ffi::ErrorCode as E;
215    'ret: loop { for retry in 0.. { match (||{
216      let mut trans = self.db.transaction().context("start transaction")?;
217      let r = f(&mut trans)?;
218      trans.commit().context("commit")?;
219      Ok::<_,AE>(r)
220    })() {
221      Err(e) => match e.downcast_ref() {
222        Some(RE::SqliteFailure(ffie, _))
223          if matches!(ffie.code, E::DatabaseBusy | E::DatabaseLocked)
224          => {
225            if retry >= MAX_RETRIES {
226              throw!(e.context("too many retries"));
227            }
228            continue;
229          },
230        _ => throw!(e),
231      }
232      Ok(r) => break 'ret r,
233    } } }
234  }
235
236  #[throws(AE)]
237  pub fn invalidate<T>(&self, key: &T::Key)
238  where T: Cacheable,
239  {
240    let keys = T::key_col_values(key).context("get keys as params")?;
241    let keys = &keys[0.. T::KEY_KCOLS.len()];
242    let stmt = (||{
243      let mut stmt = String::new();
244      write!(stmt, "DELETE FROM {}", T::TABLE)?;
245      write_cond_where_cols(&mut stmt, T::KEY_KCOLS.iter().cloned())?;
246      Ok::<_,fmt::Error>(stmt)
247    })().unwrap();
248
249    (||{
250      let mut stmt = self.db.prepare_cached(&stmt).context("prepare")?;
251      stmt.execute(&*keys).context("execute delete")?;
252      Ok::<_,AE>(())
253    })()
254      .with_context(|| format!("key={:?}", key))
255      .context("invalidate cache line")?;
256  }
257
258  #[throws(AE)]
259  pub fn lookup<T,F>(&mut self, xsa: XsaNum, key: &T::Key, mut calc: F) -> T
260  where T: Cacheable,
261        F: FnMut(&T::Key) -> Result<T,AE>
262  {
263    let all_cols = || itertools::chain(T::KEY_KCOLS, T::KEY_VCOLS).cloned();
264
265    let stmt = (||{
266      let mut stmt = String::new();
267      write!(stmt, "SELECT json FROM {}", T::TABLE)?;
268      write_cond_where_cols(&mut stmt, &mut all_cols())?;
269      Ok::<_,fmt::Error>(stmt)
270    })().unwrap();
271
272    let got = (||{
273      let keys = T::key_col_values(key).context("get keys as params")?;
274      assert_eq!( keys.len(), all_cols().count(), "{:?}", key );
275
276      let (data, refresh_token) = (||{
277        let refresh_token = self.cfg.want_refresh(T::REFRESH, Some(xsa));
278        if refresh_token.is_some() { return Ok((None, refresh_token)); }
279
280        let mut stmt = self.db.prepare_cached(&stmt).context("prepare")?;
281        let mut data = stmt.query(keys.as_slice()).context("query")?;
282        let data = data.next().context("get query row")?;
283        let data = data.map(|row| {
284          let data = row.get_ref(0).context("get result json column")?;
285          let data = match data {
286            ValueRef::Text(s) => s,
287            x => throw!(anyhow!("got wrong type, expected Text {:?}", &x)),
288          };
289          let data = str::from_utf8(data).context("got non-UTF-8 json")?;
290          let data = serde_json::from_str(data).context("parse JSON")?;
291          Ok::<T,AE>(data)
292        })
293          .transpose()?;
294        Ok::<(Option<T>, Option<RefreshingToken>),AE>((data, refresh_token))
295      })()
296        .with_context(|| stmt.clone())
297        .context("look up in cache")?;
298
299      let hit = data.is_some();
300      let data = data.map(Ok).unwrap_or_else(|| {
301        let data = calc(key).context("calculate, after cache miss")?;
302
303        let stmt = (||{
304          let mut stmt = String::new();
305          write!(stmt, "INSERT OR REPLACE INTO {} (", T::TABLE)?;
306          for k in all_cols() { write!(stmt, "{},", k)?; }
307          write!(stmt, "json) VALUES (")?;
308          for _ in all_cols() { write!(stmt, "?,")?; }
309          write!(stmt, "?)")?;
310          Ok::<_,fmt::Error>(stmt)
311        })().unwrap();
312
313        (||{
314          let json = serde_json::to_string(&data).context("make JSON")?;
315          let mut stmt = self.db.prepare_cached(&stmt).context("prepare")?;
316          let mut keys = keys;
317          keys.push(&json);
318          stmt.execute(&*keys).context("execute insert")?;
319          Ok::<_,AE>(())
320        })()
321          .with_context(|| stmt.clone())
322          .context("save in cache")?;
323
324        self.did_refresh(refresh_token);
325
326        Ok::<T,AE>(data)
327      })?;
328      Ok::<_,AE>((data, hit))
329    })();
330
331    trace!("lookup {} key={:?} {} got={:?}",
332           T::TABLE, key,
333           match got {
334             Err(_)        => "error",
335             Ok((_,false)) => "miss",
336             Ok((_,true )) => "hit",
337           },
338           &got);
339
340    got
341      .with_context(|| format!("key={:?}", key))
342      .context("lookup in cache db")?
343      .0
344  }
345}
xensec_internal_tools/cache.rs

xensec_internal_tools/
cache.rs
