Hi I’m executing a long task using the rust driver on my MongoDB atlas cluster. I was getting the “CursorNotFound” error, so I modified my code to refresh the session every 5 minutes, but now I’m getting this error “CMD_NOT_ALLOWED: refreshSessions” instead, can someone help on this? what am I doing wrong?
fn get_session_id(session: &mut ClientSession) -> Result<Bson, &str> {
match session.id().get("id") {
Some(id) => {
debug!("session id: {:?}", id);
Ok(id.clone())
}
None => {
error!("no session id");
Err("no session id")
}
}
}
pub async fn some_big_calculation(client: &Client) -> Result<(), Box<dyn std::error::Error>> {
// Get a handle to a database.
let database = client.database("database");
// List the names of the collections in that database.
for collection_name in database.list_collection_names(None).await? {
info!("collection_name: {}", collection_name);
let collection: Collection<Document> = database.collection(&collection_name);
let mut current_session = client.start_session(None).await?;
// Query the documents in the collection with a filter and an option.
let filter = doc! { "some_field": { "$exists": false } };
let find_options = FindOptions::builder().sort(doc! { "_id": -1 }).build();
let mut cursor = collection
.find_with_session(filter, find_options, &mut current_session)
.await?;
let mut last_id: i64 = 0;
if let Some(last_document) = cursor.with_session(&mut current_session).next().await {
let document = last_document?;
last_id = document.get_i64("_id")?;
//do some calculation with the first document here
// the first document is a special case
} else {
error!("Cursor not found, collection_name: {}", collection_name);
continue;
}
let wait_time = Duration::minutes(5);
let mut start = Instant::now();
let session_id = get_session_id(&mut current_session)?;
// Iterate over the results of the cursor.
// It is the previous document because we are ordering in descendent order by date/id
while let Some(previous_document) = cursor.with_session(&mut current_session).next().await {
let document = previous_document?;
let previous_id = document.get_i64("_id")?;
let some_field = expensive_calculation();
let filter = doc! { "_id" : last_id };
let update = doc! {"$set" : { "some_field": some_field}};
let update_result = collection.update_one(filter, update, None).await?;
info!(
"collection: {}, _id: {}, update_result: {:?}",
collection_name, last_id, update_result
);
last_id = previous_id;
// Check if more than 5 minutes have passed since the last refresh
match wait_time.checked_sub(start.elapsed()) > Some(0.seconds()) {
true => {
debug!(
"remaining time: {:?}",
wait_time.checked_sub(start.elapsed())
)
}
false => {
info!("5 min passed, refreshing session");
start = Instant::now();
let r = database
.run_command(doc! { "refreshSessions": [ {"id": &session_id}] }, None)
.await?;
info!("{:?}", r);
}
}
}
}
Ok(())
}
And the problem is that if I don’t try to refresh the session I always get this other error:
Error { kind: CommandError(CommandError { code: 43, code_name: “CursorNotFound”, message: “cursor id 7688697219134251972 not found”, labels: [] }), labels: [] }