mocra 0.3.0

A distributed, event-driven crawling and data collection framework
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

use async_trait::async_trait;
use dashmap::DashMap;
use dashmap::mapref::entry::Entry;

use super::types::{DagError, DagFencingStore, TaskPayload};

#[derive(Default)]
pub struct InMemoryDagFencingStore {
    latest_token_by_resource: DashMap<String, u64>,
}

#[async_trait]
impl DagFencingStore for InMemoryDagFencingStore {
    async fn commit(
        &self,
        resource: &str,
        token: u64,
        _node_id: &str,
        _payload: &TaskPayload,
    ) -> Result<(), DagError> {
        match self.latest_token_by_resource.entry(resource.to_string()) {
            Entry::Occupied(mut slot) => {
                if token <= *slot.get() {
                    return Err(DagError::FencingTokenRejected {
                        resource: resource.to_string(),
                        token,
                        reason: format!(
                            "stale token rejected by in-memory fencing store: latest_token={}",
                            slot.get()
                        ),
                    });
                }
                slot.insert(token);
                Ok(())
            }
            Entry::Vacant(slot) => {
                slot.insert(token);
                Ok(())
            }
        }
    }
}