Skip to content

chore(function-autoscaler): remove obsolete Cassandra state #1012

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
borao wants to merge 3 commits into
base: main
Choose a base branch
from
Open

chore(function-autoscaler): remove obsolete Cassandra state #1012

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2943160
chore(function-autoscaler): remove obsolete Cassandra state
borao Aug 19, 2026
56581f6
refactor(function-autoscaler): propagate Cassandra insert context
borao Aug 20, 2026
e8e7a8d
Merge branch 'main' into chore/function-autoscaler/remove-prediction-…
borao Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions migrations/cassandra/keyspaces/nvcf_autoscaler/04_drop_obsolete_function_tables.up.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
DROP TABLE IF EXISTS nvcf_autoscaler.recently_invoked_functions_history;
DROP TABLE IF EXISTS nvcf_autoscaler.running_functions_without_invocations;
DROP TABLE IF EXISTS nvcf_autoscaler.running_functions_without_invocations_history;
215 changes: 19 additions & 196 deletions ...ntrol-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ use crate::models::{
ActiveFunction, ActiveFunctionDetails, DistributedLock, DistributedLockResult, NodeHealth,
};
use crate::secrets::secrets_config::CassandraSslCertificates;
use anyhow::Result;
use anyhow::{Context, Result};
use async_trait::async_trait;
use base64::Engine;
use chrono::Utc;
Expand Down Expand Up @@ -381,75 +381,21 @@ impl CassandraServiceManager {
}
}

#[tracing::instrument(skip(self, function), fields(function_id = %function.function_id, function_version_id = %function.function_version_id))]
pub async fn insert_to_active_functions(
&self,
function: &ActiveFunctionDetails,
table: ActiveFunctionTable,
) -> Result<()> {
let session = self.get_session().await?;

let stmt = match table {
ActiveFunctionTable::RecentlyInvokedFunctions => {
get_stmt_insert_to_recently_invoked_functions(
&self.config.keyspace,
self.config.recently_invoked_ttl_seconds,
)
}
ActiveFunctionTable::RunningFunctionsWithoutInvocations => {
get_stmt_insert_to_running_functions_without_invocations(
&self.config.keyspace,
self.config.recently_invoked_ttl_seconds,
)
}
};

let nca_id = function.nca_id_or_nil();
with_cassandra_timing("insert_to_active_functions", || async {
let mut prepared_statement = session.prepare(stmt).await?;
prepared_statement.set_is_idempotent(true);
session
.execute_unpaged(
&prepared_statement,
(
&function.function_id,
&function.function_version_id,
&nca_id,
&function.last_updated_at,
),
)
.await?;
Ok(())
})
.await
}

// Not instrumented: return value is Vec<ActiveFunction> and would be captured in the span (large debug output).
pub async fn get_active_functions_with_token_range(
&self,
token_range: &[i64],
page_size: i32,
table: ActiveFunctionTable,
) -> Result<Vec<ActiveFunction>> {
let session = self.get_session().await?;
let stmt = match table {
ActiveFunctionTable::RecentlyInvokedFunctions => {
get_select_recently_invoked_functions_in_token_range_stmt(&self.config.keyspace)
}
ActiveFunctionTable::RunningFunctionsWithoutInvocations => {
get_select_running_functions_without_invocations_in_token_range_stmt(
&self.config.keyspace,
)
}
};
let stmt = get_select_recently_invoked_functions_in_token_range_stmt(&self.config.keyspace);

with_cassandra_timing("get_active_functions_with_token_range", || async {
let mut prepared_statement = session.prepare(stmt).await?;
prepared_statement.set_page_size(page_size);
// Use LOCAL_QUORUM to match the QUORUM write consistency used in add_new_active_functions_batch.
// LOCAL_ONE (execution profile default) can read from a replica that hasn't received a
// recent QUORUM write, causing a second pod to see a function as new and overwrite the
// history row with num_workers=-1 even after a prior pod already wrote it.
// recent write, causing a second pod to see the function as new.
prepared_statement.set_consistency(Consistency::LocalQuorum);
let mut results = Vec::new();
let token_range_min = token_range[0];
Expand Down Expand Up @@ -479,7 +425,7 @@ impl CassandraServiceManager {

/// Upserts a function into recently_invoked_functions with a fresh TTL.
/// Called by the scaling loop when desired_instance_count > 0 to keep the
/// function alive in the active set without touching the history table.
/// function alive in the active set.
#[tracing::instrument(skip(self))]
pub async fn refresh_active_function_ttl(&self, function: &ActiveFunction) -> Result<()> {
let session = self.get_session().await?;
Expand Down Expand Up @@ -510,54 +456,40 @@ impl CassandraServiceManager {
pub async fn add_new_active_functions_batch(
&self,
functions: &[ActiveFunctionDetails],
table: ActiveFunctionTable,
) -> Result<()> {
if functions.is_empty() {
return Ok(());
}

let session = self.get_session().await?;
let stmt_active_function = match table {
ActiveFunctionTable::RecentlyInvokedFunctions => {
get_stmt_insert_to_recently_invoked_functions(
&self.config.keyspace,
self.config.recently_invoked_ttl_seconds,
)
}
ActiveFunctionTable::RunningFunctionsWithoutInvocations => {
get_stmt_insert_to_running_functions_without_invocations(
&self.config.keyspace,
self.config.recently_invoked_ttl_seconds,
)
}
};
let mut prepared_active_function = session.prepare(stmt_active_function).await?;
prepared_active_function.set_consistency(scylla::statement::Consistency::Quorum);
prepared_active_function.set_is_idempotent(true);
let stmt = get_stmt_insert_to_recently_invoked_functions(
&self.config.keyspace,
self.config.recently_invoked_ttl_seconds,
);
let mut prepared = session.prepare(stmt).await?;
prepared.set_consistency(Consistency::Quorum);
prepared.set_is_idempotent(true);
Comment thread
coderabbitai[bot] marked this conversation as resolved.

with_cassandra_timing("add_new_active_functions_batch", || async {
execute_chunked(functions, 200, |function| {
let session = session.clone();
let prepared_active_function = prepared_active_function.clone();
let prepared = prepared.clone();
let function_id = function.function_id;
let function_version_id = function.function_version_id;
let nca_id = function.nca_id_or_nil();
let nca_id = function.nca_id.clone().unwrap_or_default();
let last_updated_at = function.last_updated_at;
async move {
session
.execute_unpaged(
&prepared_active_function,
&prepared,
(&function_id, &function_version_id, &nca_id, last_updated_at),
)
.await
.map_err(|e| {
tracing::error!(
"Failed to insert function {}:{} to Cassandra: {}",
function_id,
function_version_id,
e
);
anyhow::Error::from(e)
.with_context(|| {
format!(
"inserting function {}:{} into recently_invoked_functions",
function_id, function_version_id
)
})
}
})
Expand All @@ -568,52 +500,6 @@ impl CassandraServiceManager {
Ok(())
}

#[tracing::instrument(skip(self))]
pub async fn delete_active_function(
&self,
function_id: &Uuid,
function_version_id: &Uuid,
table: ActiveFunctionTable,
) -> Result<()> {
let session = self.get_session().await?;
let stmt_active_function = match table {
ActiveFunctionTable::RecentlyInvokedFunctions => {
get_delete_recently_invoked_function_stmt(&self.config.keyspace)
}
ActiveFunctionTable::RunningFunctionsWithoutInvocations => {
get_delete_running_function_without_invocations_stmt(&self.config.keyspace)
}
};
let mut prepared = session.prepare(stmt_active_function).await?;
prepared.set_consistency(scylla::statement::Consistency::Quorum);
prepared.set_is_idempotent(true);
let result = with_cassandra_timing("delete_active_function", || async {
session
.execute_unpaged(&prepared, (function_id, function_version_id))
.await
})
.await;
match result {
Ok(_) => {
tracing::debug!(
"Successfully deleted function {}:{} from Cassandra",
function_id,
function_version_id
);
}
Err(e) => {
tracing::error!(
"Failed to delete function {}:{} from Cassandra: {}",
function_id,
function_version_id,
e
);
return Err(e.into());
}
}
Ok(())
}

// Returns true if the lock was acquired, false if it was already held by another node
#[tracing::instrument(skip(self))]
pub async fn put_lock(&self, lock: &DistributedLock, ttl_seconds: i32) -> Result<bool> {
Expand Down Expand Up @@ -973,16 +859,6 @@ mod tests {
}
}

fn create_test_active_function_details() -> ActiveFunctionDetails {
ActiveFunctionDetails {
function_id: Uuid::new_v4(),
function_version_id: Uuid::new_v4(),
nca_id: Some("test-nca-id".to_string()),
num_workers: Some(1),
last_updated_at: Some(Utc::now()),
}
}

fn create_test_lock() -> DistributedLock {
DistributedLock {
lock_name: "test_lock".to_string(),
Expand Down Expand Up @@ -1012,59 +888,6 @@ mod tests {
assert!(result.is_ok());
}

#[tokio::test]
#[ignore = "Requires running Cassandra"]
async fn test_active_function_operations() {
let settings = create_test_settings().await;
let secrets_path = get_test_secrets_path();
let secrets_watcher = Arc::new(
SecretFileWatcher::new(Path::new(&secrets_path))
.await
.unwrap(),
);
let manager = CassandraServiceManager::new(&settings, secrets_watcher)
.await
.unwrap();

for table_type in [
ActiveFunctionTable::RecentlyInvokedFunctions,
ActiveFunctionTable::RunningFunctionsWithoutInvocations,
] {
let function = create_test_active_function_details();
manager
.add_new_active_functions_batch(std::slice::from_ref(&function), table_type)
.await
.unwrap();

let token_range = CASSANDRA_TOKEN_RANGE;
let functions = manager
.get_active_functions_with_token_range(&token_range, 100, table_type)
.await
.unwrap();
assert!(functions.iter().any(|active| {
active.function_id == function.function_id
&& active.function_version_id == function.function_version_id
}));

manager
.delete_active_function(
&function.function_id,
&function.function_version_id,
table_type,
)
.await
.unwrap();
let functions = manager
.get_active_functions_with_token_range(&token_range, 100, table_type)
.await
.unwrap();
assert!(!functions.iter().any(|active| {
active.function_id == function.function_id
&& active.function_version_id == function.function_version_id
}));
}
}

#[tokio::test]
#[ignore = "Requires running Cassandra"]
async fn test_lock_operations() {
Expand Down
1 change: 1 addition & 0 deletions ...trol-plane-services/function-autoscaler/crates/server/src/cassandra/cassandra_settings.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ const DEFAULT_SERIAL_CONSISTENCY: &str = "LOCAL_SERIAL";
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_millis(10000); // 10 seconds
const DEFAULT_MAX_RETRY_COUNT: u32 = 3;
const DEFAULT_RETRY_INTERVAL: Duration = Duration::from_millis(1000); // 1 second

#[serde_as]
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default)]
Expand Down
Loading
Loading