liminal-server 0.25.4

Standalone server for the liminal messaging bus
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

//! The agent draft store against a real durable journal.

use super::super::waiting::{
    AgentDelivery, WaitingLimits,
    tests::{DiskFixture, create_agent, held, keeper, placer, pull_limits, store},
};
use super::super::{DraftExecutor, tests::Issuer};
use super::*;
use liminal::channel::{ChannelConfig, ChannelHandle, ChannelMode, Schema};
use lys_pass::drafts::execute_once;
use serde_json::json;
use std::collections::BTreeMap;
use std::num::NonZeroUsize;

// Two workers let sockets progress while a registry call waits for HTTP.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn held_agent_act_clears_waiting_and_publishes_one_real_durable_inbox_outcome()
-> Result<(), Box<dyn std::error::Error>> {
    let issuer = Issuer::new(vec![(200, json!({})); 8])?;
    let disk = DiskFixture::new()?;
    let placement = placer(&issuer)?;
    let owners = disk.owners(placement.clone())?;
    let lifecycle = disk.lifecycle(placement)?;
    let created = create_agent("builder-create")?;
    held(
        &keeper(
            owners.clone(),
            lifecycle.clone(),
            "fixture-owner",
            "person",
            &created,
        )?,
        &created,
    )?;
    let mut first = DraftExecutor {
        store: store(&owners)?,
        limits: pull_limits()?,
        at_ms: 2000,
    };
    execute_once(&created, &mut first)?;
    let draft = super::super::admin::tests::fixture_child_request(owners.clone(), lifecycle)?;
    let state = owners
        .agents()
        .retained()
        .value("agents.builder.state")?
        .ok_or("waiting state missing")?;
    assert_eq!(state["state"], "waiting");
    assert_eq!(state["words"], draft.words);
    assert_eq!(state["draft"], draft.id);
    assert_eq!(state["since"], 1000);
    let inbox = ChannelHandle::new_durable(
        ChannelConfig::new(
            "agents.builder.inbox".into(),
            Schema::new(json!({}))?,
            ChannelMode::Durable,
        ),
        disk.store,
    )?;
    let waiting = WaitingLimits {
        drafts: NonZeroUsize::new(8).ok_or("zero bound")?,
        words_bytes: NonZeroUsize::new(4096).ok_or("zero bound")?,
        identifier_bytes: NonZeroUsize::new(256).ok_or("zero bound")?,
    };
    let delivery = AgentDelivery::new(
        &owners,
        BTreeMap::from([("builder".into(), inbox.clone())]),
        "fixture-exchange".into(),
        waiting,
    )?;
    let mut executor = DraftExecutor {
        store: AgentDraftStore::new(&owners, delivery),
        limits: pull_limits()?,
        at_ms: 3000,
    };
    let receipt = execute_once(&draft, &mut executor)?;
    executor.store.deliver_outcome(&draft, &receipt)?;
    executor.store.deliver_outcome(&draft, &receipt)?;
    assert_eq!(inbox.durable_positions()?[0], 1);
    let state = owners
        .agents()
        .retained()
        .value("agents.builder.state")?
        .ok_or("terminal state missing")?;
    assert_eq!(state["state"], "gone");
    assert_eq!(state["outcome"], "executed");
    assert_eq!(
        owners
            .agents()
            .registry()
            .record("child")?
            .transitions
            .len(),
        1
    );
    let message = owners
        .journal()
        .read(|snapshot| serde_json::to_value(&snapshot.deliveries[&draft.id]))??;
    assert_eq!(message["message"]["draft"], draft.id);
    assert_eq!(message["message"]["meta"]["from"], "fixture-exchange");
    issuer.finish()
}