appcore-gateway 1.0.2-rc

Multi-tenant Gateway capability for the AppCore Runtime.
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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

// =============================================================================
//        #######
//     ###       ###     F: socket_ownership.rs
//    ##   ## ##   ##    P: AppCore-Runtime
//         ## ##
//                       C: 2026/08/26 00:00:00 by dnettoRaw
//    ##   ## ##   ##    U: 2026/08/26 00:00:00 by dnettoRaw
//      ###########      S: 2.0.0-beta.1
// =============================================================================

//! Atomic ordering between shared HA ownership and local socket directories.

use crate::config::MAX_GATEWAY_CONNECTIONS;
use crate::{
    ClientConnection, GatewaySession, GatewayState, GatewayWorkerRegistration, WorkerConnection,
};
use appcore_contracts::InstallationId;
use appcore_types::{CapabilityName, ClusterId, CoreId, TenantId};
use std::time::{SystemTime, UNIX_EPOCH};

pub(crate) async fn register_worker_ownership(
    state: &GatewayState,
    tenant_id: &TenantId,
    cluster_id: &ClusterId,
    installation_id: &InstallationId,
    core_id: &CoreId,
    connection: &WorkerConnection,
    capabilities: Vec<CapabilityName>,
) -> Result<bool, &'static str> {
    if let Some(coordinator) = state.ha_coordinator() {
        let registration = GatewayWorkerRegistration::new(
            installation_id.clone(),
            core_id.clone(),
            connection.generation(),
            capabilities.clone(),
        )
        .map_err(|_| "invalid HA worker ownership")?;
        coordinator
            .register_worker(tenant_id, cluster_id, registration, now_ms())
            .await
            .map_err(|_| "HA registry rejected worker ownership")?;
    }
    let local = {
        let _admission = state.lock_connection_admission();
        if state.connection_count() >= MAX_GATEWAY_CONNECTIONS {
            Err("global connection limit")
        } else {
            state
                .tenant_partition_or_insert(tenant_id)
                .map_err(|_| "tenant limit")
                .and_then(|tenant| {
                    let mut tenant = tenant.write();
                    let replaced = tenant.get_worker(installation_id, core_id).is_some();
                    tenant
                        .add_worker(connection.clone(), capabilities)
                        .map(|()| replaced)
                        .map_err(|_| "worker limit")
                })
        }
    };
    if local.is_err() {
        remove_worker_ownership(state, tenant_id, installation_id, core_id, connection).await;
    }
    local
}

pub(crate) async fn remove_worker_ownership(
    state: &GatewayState,
    tenant_id: &TenantId,
    installation_id: &InstallationId,
    core_id: &CoreId,
    connection: &WorkerConnection,
) {
    if let Some(coordinator) = state.ha_coordinator() {
        let _ = coordinator
            .remove_worker(tenant_id, installation_id, core_id, connection.generation())
            .await;
    }
}

pub(crate) async fn register_client_ownership(
    state: &GatewayState,
    tenant_id: &TenantId,
    cluster_id: &ClusterId,
    connection: &ClientConnection,
    session: GatewaySession,
) -> Result<(), &'static str> {
    if let Some(coordinator) = state.ha_coordinator() {
        coordinator
            .register_session(
                tenant_id,
                cluster_id,
                session.session_id.clone(),
                session.expires_at_ms,
                now_ms(),
            )
            .await
            .map_err(|_| "HA registry rejected session ownership")?;
    }
    let local = {
        let _admission = state.lock_connection_admission();
        if state.connection_count() >= MAX_GATEWAY_CONNECTIONS {
            Err("global connection limit")
        } else {
            state
                .tenant_partition_or_insert(tenant_id)
                .map_err(|_| "tenant limit")
                .and_then(|tenant| {
                    let mut tenant = tenant.write();
                    tenant
                        .try_add_client(connection.clone())
                        .map_err(|_| "client limit")?;
                    tenant.sessions.insert(session.session_id.clone(), session);
                    Ok(())
                })
        }
    };
    if local.is_err() {
        remove_client_ownership(state, tenant_id, &connection.session_id).await;
    }
    local
}

pub(crate) async fn remove_client_ownership(
    state: &GatewayState,
    tenant_id: &TenantId,
    session_id: &str,
) {
    if let Some(coordinator) = state.ha_coordinator() {
        let _ = coordinator.remove_session(tenant_id, session_id).await;
    }
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| {
            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
        })
}