reifydb_engine/
subscription.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{result::Result as StdResult, sync::Arc};
5
6use reifydb_core::{
7	common::CommitVersion, interface::catalog::id::SubscriptionId, metrics::execution::ExecutionMetrics,
8	value::column::columns::Columns,
9};
10use reifydb_evaluate::stack::SymbolTable;
11use reifydb_rql::flow::flow::FlowDag;
12use reifydb_transaction::{multi::lease::VersionLeaseGuard, transaction::Transaction};
13use reifydb_value::{Result, error::Error as TypeError, params::Params, value::identity::IdentityId};
14
15use crate::engine::StandardEngine;
16
17#[derive(Debug, Clone)]
18pub struct SubscriptionContext {
19	pub id: SubscriptionId,
20	pub identity: IdentityId,
21	pub symbols: SymbolTable,
22	pub params: Params,
23}
24
25#[derive(Debug)]
26pub enum HydrateError {
27	SubscriptionNotFound,
28	UnsupportedSourceType,
29	RowCapExceeded {
30		cap: u64,
31	},
32	Engine(TypeError),
33	Internal(String),
34}
35
36impl From<TypeError> for HydrateError {
37	fn from(e: TypeError) -> Self {
38		HydrateError::Engine(e)
39	}
40}
41
42impl HydrateError {
43	pub fn is_version_evicted(&self) -> bool {
44		matches!(self, HydrateError::Engine(e) if e.0.code == "TXN_012")
45	}
46
47	pub fn wire_code(&self) -> &'static str {
48		match self {
49			Self::SubscriptionNotFound => "HYDRATION_FAILED",
50			Self::UnsupportedSourceType => "HYDRATION_UNSUPPORTED_SOURCE",
51			Self::RowCapExceeded {
52				..
53			} => "HYDRATION_TOO_LARGE",
54			Self::Engine(_) => {
55				if self.is_version_evicted() {
56					"HYDRATION_VERSION_EVICTED"
57				} else {
58					"HYDRATION_FAILED"
59				}
60			}
61			Self::Internal(_) => "HYDRATION_FAILED",
62		}
63	}
64
65	pub fn wire_message(&self, rql: &str, cap: u64) -> String {
66		match self {
67			Self::SubscriptionNotFound => "Subscription not found at hydration time".to_string(),
68			Self::UnsupportedSourceType => "hydration is not supported for SourceSeries / SourceInlineData; use WITH { hydration: { enabled: false } } to subscribe without it".to_string(),
69			Self::RowCapExceeded { .. } => format!(
70				"Hydration exceeds subscribe.max_hydration_rows={}; add `TAKE N` upstream, lower with WITH {{ hydration: {{ max_rows: ... }} }}, or disable with WITH {{ hydration: {{ enabled: false }} }}. Query: {}",
71				cap, rql
72			),
73			Self::Engine(e) => {
74				if self.is_version_evicted() {
75					e.0.message.clone()
76				} else {
77					e.to_string()
78				}
79			}
80			Self::Internal(s) => s.clone(),
81		}
82	}
83}
84
85#[derive(Debug)]
86pub struct HydrateOutcome {
87	pub version: CommitVersion,
88	pub batches: Vec<Columns>,
89	pub metrics: ExecutionMetrics,
90}
91
92pub trait SubscriptionService: Send + Sync {
93	fn next_id(&self) -> SubscriptionId;
94
95	fn register_subscription(
96		&self,
97		flow_dag: FlowDag,
98		column_names: Vec<String>,
99		hydration_enabled: bool,
100		ctx: SubscriptionContext,
101		txn: &mut Transaction<'_>,
102	) -> Result<()>;
103
104	fn unregister_subscription(&self, id: &SubscriptionId) -> Result<()>;
105
106	fn hydrate(
107		&self,
108		sub_id: SubscriptionId,
109		engine: &StandardEngine,
110		identity: IdentityId,
111		lease: VersionLeaseGuard,
112		max_rows: u64,
113	) -> StdResult<HydrateOutcome, HydrateError>;
114}
115
116pub type SubscriptionServiceRef = Arc<dyn SubscriptionService>;
reifydb_engine/subscription.rs

reifydb_engine/
subscription.rs
