reifydb-store-multi 0.8.1

Multi-version storage for OLTP operations with MVCC support
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

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use std::sync::Arc;

use reifydb_core::{
	actors::version_epoch::VersionEpochMessage as Message,
	interface::catalog::config::{ConfigKey, GetConfig},
};
use reifydb_runtime::{
	actor::{
		context::Context,
		mailbox::ActorRef,
		system::{ActorConfig, ActorSpawner},
		timers::TimerHandle,
		traits::{Actor as ActorTrait, Directive},
	},
	version_epoch::VersionEpoch,
};
use reifydb_value::value::datetime::DateTime;
use tracing::{debug, trace};

use super::EpochSource;

pub struct ActorState {
	_timer_handle: Option<TimerHandle>,
}

pub struct Actor<S: EpochSource> {
	epoch: VersionEpoch,
	source: S,
	config: Arc<dyn GetConfig>,
}

impl<S: EpochSource> Actor<S> {
	pub fn new(epoch: VersionEpoch, source: S, config: Arc<dyn GetConfig>) -> Self {
		Self {
			epoch,
			source,
			config,
		}
	}

	pub fn spawn(
		spawner: &ActorSpawner,
		epoch: VersionEpoch,
		source: S,
		config: Arc<dyn GetConfig>,
	) -> ActorRef<Message> {
		let actor = Self::new(epoch, source, config);
		spawner.spawn_coordination("version-epoch-sampler", actor).actor_ref().clone()
	}

	fn sample(&self) {
		let now_nanos = self.source.now_nanos();
		let Some(version) = self.source.current_version() else {
			trace!("Version epoch sampler: no committed version yet, skipping sample");
			return;
		};
		self.epoch.record(now_nanos, version.0);
	}
}

impl<S: EpochSource> ActorTrait for Actor<S> {
	type State = ActorState;
	type Message = Message;

	fn init(&self, ctx: &Context<Message>) -> ActorState {
		debug!("Version epoch sampler started");
		let interval = self.config.get_config_duration(ConfigKey::VersionEpochSampleInterval);
		let timer_handle = ctx.schedule_tick(interval, |nanos| Message::Tick(DateTime::from_nanos(nanos)));
		self.sample();
		ActorState {
			_timer_handle: Some(timer_handle),
		}
	}

	fn handle(&self, _state: &mut ActorState, msg: Message, ctx: &Context<Message>) -> Directive {
		if ctx.is_cancelled() {
			return Directive::Stop;
		}

		match msg {
			Message::Tick(_) => self.sample(),
			Message::Shutdown => {
				debug!("Version epoch sampler shutting down");
				return Directive::Stop;
			}
		}

		Directive::Yield
	}

	fn post_stop(&self) {
		debug!("Version epoch sampler stopped");
	}

	fn config(&self) -> ActorConfig {
		ActorConfig::new().mailbox_capacity(64)
	}
}

pub fn spawn_version_epoch_sampler<S: EpochSource>(
	epoch: VersionEpoch,
	spawner: ActorSpawner,
	source: S,
	config: Arc<dyn GetConfig>,
) -> ActorRef<Message> {
	Actor::spawn(&spawner, epoch, source, config)
}