folk-api 0.2.5

Plugin contract for the Folk PHP application server
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

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

use anyhow::Result;
use async_trait::async_trait;
use folk_api::{Executor, Plugin, PluginContext, ServerPlugin, ServerPluginWrapper};
use tokio::sync::watch;
use tokio::time::timeout;

struct EchoPlugin;

#[async_trait]
impl ServerPlugin for EchoPlugin {
    fn name(&self) -> &'static str {
        "echo"
    }

    async fn run(&self, mut ctx: PluginContext) -> Result<()> {
        if let Err(e) = ctx.shutdown.changed().await {
            tracing::error!(error = %e, "shutdown sender dropped unexpectedly");
        }
        Ok(())
    }
}

struct NoopExecutor;

#[async_trait]
impl Executor for NoopExecutor {
    async fn execute_method(&self, _: &str, _: bytes::Bytes) -> Result<bytes::Bytes> {
        unimplemented!("not used in this test")
    }
}

#[tokio::test]
async fn server_plugin_wrapper_boots_and_shuts_down() {
    let mut wrapper = ServerPluginWrapper::new(EchoPlugin);

    let (tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();
    tokio::time::sleep(Duration::from_millis(50)).await;
    tx.send(true).unwrap();

    wrapper.shutdown().await.unwrap();
}

/// Regression test for issue #54: boot() must return an error when called
/// a second time while the plugin is already running.
#[tokio::test]
async fn server_plugin_wrapper_double_boot_returns_error() {
    let mut wrapper = ServerPluginWrapper::new(EchoPlugin);
    let (_tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx.clone(),
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };
    let ctx2 = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();
    let err = wrapper.boot(ctx2).await;
    assert!(err.is_err(), "second boot() must return Err");
    assert!(
        err.unwrap_err().to_string().contains("already running"),
        "error message must mention 'already running'"
    );
}

/// Regression test for issue #55: shutdown() must not deadlock when a plugin
/// task ignores ctx.shutdown and never exits on its own.
#[tokio::test]
async fn server_plugin_wrapper_shutdown_aborts_stuck_task() {
    struct StubbornPlugin;

    #[async_trait]
    impl ServerPlugin for StubbornPlugin {
        fn name(&self) -> &'static str {
            "stubborn"
        }

        async fn run(&self, _ctx: PluginContext) -> Result<()> {
            // Ignores ctx.shutdown — would block forever without abort().
            std::future::pending::<()>().await;
            Ok(())
        }
    }

    let mut wrapper = ServerPluginWrapper::new(StubbornPlugin);
    let (_tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();

    // shutdown() must return quickly; a deadlock would cause a test timeout.
    let result = timeout(Duration::from_secs(5), wrapper.shutdown()).await;
    assert!(result.is_ok(), "shutdown() deadlocked");
    result.unwrap().unwrap();
}