scopedb-client 0.2.2

A Rust client for ScopeDB
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

// Copyright 2024 ScopeDB, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

mod common;

use std::time::Duration;

use scopedb_client::AppendDeliveryOutcome;
use scopedb_client::AppendFailurePolicy;
use scopedb_client::AppendStream;
use uuid::Uuid;

fn track(stream: &AppendStream, request: u64) -> bool {
    match stream.try_send(&serde_json::json!({
        "event_id": Uuid::now_v7(),
        "occurred_at": jiff::Timestamp::now(),
        "name": "request.completed",
        "attributes": {"request": request, "status": 200},
    })) {
        // This means local admission only, not a remote commit.
        Ok(()) => true,
        Err(error) => {
            // Route this diagnostic to another sink, never back into `stream`.
            eprintln!("telemetry row dropped locally: {error}");
            false
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .batch_bytes(1024 * 1024)
        .flush_interval(Duration::from_secs(1))
        .max_pending_bytes(32 * 1024 * 1024)
        .max_in_flight_requests(2)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}