gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-05 21:07:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(time): migrate audit and notify timestamps to jiff (#5707)

Browse Source 
* refactor(time): migrate audit and notify timestamps to jiff

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): initialize heal walk decode error

Co-Authored-By: heihutu <heihutu@gmail.com>

* refactor(targets): parse MySQL event time with jiff

Preserve MySQL DATETIME(6) wall-time formatting for RFC3339 eventTime values while removing the direct chrono dependency from rustfs-targets.

Co-Authored-By: heihutu <heihutu@gmail.com>

* chore(deps): prune unused workspace dependencies

Apply cargo shear --fix to remove unused path-clean and s3select-api tempfile entries after the scoped jiff migration.

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): remove duplicate heal walk decode error init

Remove the duplicate decode_error field from the heal walk test collector initializer so lib-test clippy compiles on CI.

Co-Authored-By: heihutu <heihutu@gmail.com>

* refactor(policy): emit OPA timestamps with jiff

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-04 23:20:44 +08:00
committed by GitHub
parent b14805af47
commit 510b0350d6
16 changed files with 206 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+4 -5
View File
@@ -9135,11 +9135,11 @@ name = "rustfs-audit"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"chrono",
"const-str",
"futures",
"hashbrown 0.17.1",
"hotpath",
"jiff",
"metrics",
"rustfs-config",
"rustfs-s3-types",
@@ -9737,11 +9737,11 @@ dependencies = [
"arc-swap",
"async-trait",
"axum",
"chrono",
"criterion",
"form_urlencoded",
"hashbrown 0.17.1",
"hotpath",
"jiff",
"metrics",
"percent-encoding",
"quick-xml",
@@ -9862,10 +9862,10 @@ version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"base64-simd",
"chrono",
"futures",
"hotpath",
"ipnetwork",
"jiff",
"jsonwebtoken 11.0.0",
"moka",
"pollster",
@@ -10091,7 +10091,6 @@ dependencies = [
"s3s",
"serde_json",
"serial_test",
"tempfile",
"thiserror 2.0.19",
"tokio",
"tokio-util",
@@ -10205,7 +10204,6 @@ dependencies = [
"arc-swap",
"async-nats",
"async-trait",
"chrono",
"criterion",
"deadpool-postgres",
"futures-util",
@@ -10213,6 +10211,7 @@ dependencies = [
"hotpath",
"hyper",
"hyper-rustls",
"jiff",
"lapin",
"libc",
"metrics",
Cargo.toml
-1
View File
@@ -274,7 +274,6 @@ num_cpus = { version = "1.17.0" }
nvml-wrapper = "0.12.1"
parking_lot = "0.12.5"
path-absolutize = "4.0.1"
path-clean = "1.0.1"
percent-encoding = "2.3.2"
pin-project-lite = "0.2.17"
pretty_assertions = "1.4.1"
crates/audit/Cargo.toml
+1 -1
View File
@@ -55,10 +55,10 @@ hotpath.workspace = true
rustfs-targets = { workspace = true }
rustfs-config = { workspace = true, features = ["audit", "server-config-model"] }
rustfs-s3-types = { workspace = true }
chrono = { workspace = true, features = ["serde"] }
const-str = { workspace = true, features = ["std", "proc"] }
futures = { workspace = true }
hashbrown = { workspace = true, features = ["serde", "rayon"] }
jiff = { workspace = true, features = ["serde"] }
metrics = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["raw_value"] }
crates/audit/src/entity.rs
+24 -5
View File
@@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use chrono::{DateTime, Utc};
use hashbrown::HashMap;
use jiff::Timestamp;
use rustfs_s3_types::EventName;
use serde::{Deserialize, Serialize};
use serde_json::Value;
@@ -151,8 +151,8 @@ pub struct AuditEntry {
pub deployment_id: Option<String>,
#[serde(rename = "siteName", skip_serializing_if = "Option::is_none")]
pub site_name: Option<String>,
#[serde(with = "chrono::serde::ts_milliseconds")]
pub time: DateTime<Utc>,
#[serde(with = "jiff::fmt::serde::timestamp::millisecond::required")]
pub time: Timestamp,
pub event: EventName,
#[serde(rename = "type", skip_serializing_if = "Option::is_none")]
pub entry_type: Option<String>,
@@ -198,7 +198,7 @@ impl AuditEntryBuilder {
pub fn new(version: impl Into<String>, event: EventName, trigger: impl Into<String>, api: ApiDetails) -> Self {
Self(AuditEntry {
version: version.into(),
time: Utc::now(),
time: Timestamp::now(),
event,
trigger: trigger.into(),
api,
@@ -232,7 +232,7 @@ impl AuditEntryBuilder {
self
}
pub fn time(mut self, time: DateTime<Utc>) -> Self {
pub fn time(mut self, time: Timestamp) -> Self {
self.0.time = time;
self
}
@@ -342,4 +342,23 @@ mod tests {
assert_eq!(value["requestID"], Value::String("req-audit-123".to_string()));
assert!(value.get("request_id").is_none(), "historical audit contract must not expose request_id");
}
#[test]
fn audit_entry_time_serializes_as_epoch_milliseconds() {
let entry = AuditEntryBuilder::new(
"1",
EventName::ObjectCreatedPut,
"s3",
ApiDetailsBuilder::new()
.name("PutObject")
.status("OK")
.status_code(200)
.build(),
)
.time(Timestamp::from_millisecond(1_711_423_698_870).expect("timestamp should be valid"))
.build();
let value = serde_json::to_value(entry).expect("audit entry should serialize");
assert_eq!(value["time"], Value::Number(1_711_423_698_870_i64.into()));
}
}
crates/audit/tests/performance_test.rs
+3 -3
View File
@@ -97,7 +97,7 @@ async fn test_audit_log_dispatch_performance() {
return; // Alternatively: assert!(false, "AuditSystem failed to start");
}
use chrono::Utc;
use jiff::Timestamp;
use rustfs_targets::EventName;
use serde_json::json;
use std::collections::HashMap;
@@ -136,7 +136,7 @@ async fn test_audit_log_dispatch_performance() {
version: "1".to_string(),
deployment_id: Some(format!("test-deployment-{id}")),
site_name: Some("test-site".to_string()),
time: Utc::now(),
time: Timestamp::now(),
event: EventName::ObjectCreatedPut,
entry_type: Some("object".to_string()),
trigger: "api".to_string(),
@@ -298,7 +298,7 @@ fn test_performance_requirements() {
for i in 0..3000 {
// Simulate event name parsing and processing
let _event_id = format!("s3:ObjectCreated:Put_{i}");
let _timestamp = chrono::Utc::now().to_rfc3339();
let _timestamp = jiff::Timestamp::now().to_string();
// Simulate basic audit entry creation overhead
let _entry_size = 512; // bytes
crates/audit/tests/system_integration_test.rs
+2 -2
View File
@@ -264,7 +264,7 @@ fn create_sample_audit_entry() -> AuditEntry {
}
fn create_sample_audit_entry_with_id(id: u32) -> AuditEntry {
use chrono::Utc;
use jiff::Timestamp;
use rustfs_targets::EventName;
use serde_json::json;
@@ -301,7 +301,7 @@ fn create_sample_audit_entry_with_id(id: u32) -> AuditEntry {
version: "1".to_string(),
deployment_id: Some(format!("test-deployment-{id}")),
site_name: Some("test-site".to_string()),
time: Utc::now(),
time: Timestamp::now(),
event: EventName::ObjectCreatedPut,
entry_type: Some("object".to_string()),
trigger: "api".to_string(),
crates/ecstore/src/set_disk/ops/heal_walk.rs
+1 -1
View File
@@ -492,8 +492,8 @@ mod tests {
batch_objects: 1000,
version_budget: 10_000,
objects: Mutex::new(Vec::new()),
decode_error: Mutex::new(None),
version_total: AtomicUsize::new(0),
decode_error: Mutex::new(None),
truncated: AtomicBool::new(false),
cancel: CancellationToken::new(),
});
crates/notify/Cargo.toml
+1 -1
View File
@@ -68,9 +68,9 @@ rustfs-targets = { workspace = true }
rustfs-utils = { workspace = true }
arc-swap = { workspace = true }
async-trait = { workspace = true }
chrono = { workspace = true, features = ["serde"] }
form_urlencoded = { workspace = true }
hashbrown = { workspace = true, features = ["serde", "rayon"] }
jiff = { workspace = true, features = ["serde"] }
percent-encoding = { workspace = true }
rayon = { workspace = true }
rustc-hash = { workspace = true }
crates/notify/examples/webhook.rs
+3 -3
View File
@@ -110,7 +110,7 @@ async fn reset_webhook_count(Query(params): Query<ResetParams>, headers: HeaderM
let reason = params.reason.unwrap_or_else(|| "Reason not provided".to_string());
println!("Reset webhook count, reason: {reason}");
let time_now = chrono::offset::Utc::now().to_string();
let time_now = jiff::Timestamp::now().to_string();
for header in headers {
let (key, value) = header;
println!("Header: {key:?}: {value:?}, time: {time_now}");
@@ -120,7 +120,7 @@ async fn reset_webhook_count(Query(params): Query<ResetParams>, headers: HeaderM
// Reset the counter to 0
WEBHOOK_COUNT.store(0, Ordering::SeqCst);
println!("Webhook count has been reset to 0.");
let time_now = chrono::offset::Utc::now().to_string();
let time_now = jiff::Timestamp::now().to_string();
Response::builder()
.header("Foo", "Bar")
.status(StatusCode::OK)
@@ -171,7 +171,7 @@ async fn receive_webhook(Json(payload): Json<Value>) -> StatusCode {
println!(
"Total webhook requests received: {} , Time: {}",
WEBHOOK_COUNT.load(Ordering::SeqCst),
chrono::offset::Utc::now()
jiff::Timestamp::now()
);
StatusCode::OK
}
crates/notify/src/event.rs
+65 -16
View File
@@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use chrono::{DateTime, SecondsFormat, Utc};
use hashbrown::HashMap;
use jiff::{Timestamp, tz::TimeZone};
use rustfs_s3_ops::is_object_removed_event;
use rustfs_s3_types::{EventName, event_schema_version};
use rustfs_utils::http::{is_encryption_metadata_key, is_internal_key};
@@ -91,8 +91,8 @@ pub struct NotifyObjectInfo {
pub content_type: Option<String>,
pub user_defined: HashMap<String, String>,
pub version_id: Option<String>,
pub mod_time: Option<DateTime<Utc>>,
pub restore_expires: Option<DateTime<Utc>>,
pub mod_time: Option<Timestamp>,
pub restore_expires: Option<Timestamp>,
pub storage_class: Option<String>,
pub transitioned_tier: Option<String>,
}
@@ -150,8 +150,11 @@ pub struct Event {
/// The AWS region where the event occurred
pub aws_region: String,
/// The time when the event occurred
#[serde(serialize_with = "serialize_event_time_millis")]
pub event_time: DateTime<Utc>,
#[serde(
serialize_with = "serialize_event_time_millis",
deserialize_with = "deserialize_event_time_millis"
)]
pub event_time: Timestamp,
/// The name of the event
pub event_name: EventName,
/// The identity of the user who triggered the event
@@ -189,13 +192,13 @@ impl Event {
user_metadata.insert("x-rustfs-object-size".to_string(), "1024".to_string());
user_metadata.insert("x-rustfs-object-etag".to_string(), "etag123".to_string());
user_metadata.insert("x-rustfs-object-version-id".to_string(), "1".to_string());
user_metadata.insert("x-request-time".to_string(), Utc::now().to_rfc3339());
user_metadata.insert("x-request-time".to_string(), format_timestamp_rfc3339_millis(&Timestamp::now()));
Event {
event_version: event_schema_version(event_name).to_string(),
event_source: "rustfs:s3".to_string(),
aws_region: "us-east-1".to_string(),
event_time: Utc::now(),
event_time: Timestamp::now(),
event_name,
user_identity: Identity {
principal_id: "rustfs".to_string(),
@@ -236,10 +239,10 @@ impl Event {
}
pub fn new(args: EventArgs) -> Self {
let event_time = Utc::now().naive_local();
let event_time = Timestamp::now();
let sequencer = match args.object.mod_time {
Some(t) => format!("{:X}", t.timestamp_nanos_opt().unwrap_or(0)),
None => format!("{:X}", event_time.and_utc().timestamp_nanos_opt().unwrap_or(0)),
Some(t) => sequencer_from_timestamp(t),
None => sequencer_from_timestamp(event_time),
};
let mut resp_elements = args.resp_elements.clone();
@@ -312,7 +315,7 @@ impl Event {
.or_else(|| args.object.transitioned_tier.clone())?;
Some(GlacierEventData {
restore_event_data: RestoreEventData {
lifecycle_restoration_expiry_time: expiry_time.to_rfc3339_opts(SecondsFormat::Millis, true),
lifecycle_restoration_expiry_time: format_timestamp_rfc3339_millis(&expiry_time),
lifecycle_restore_storage_class: storage_class,
},
})
@@ -325,7 +328,7 @@ impl Event {
event_version: event_schema_version(args.event_name).to_string(),
event_source: "rustfs:s3".to_string(),
aws_region: args.req_params.get("region").cloned().unwrap_or_default(),
event_time: event_time.and_utc(),
event_time,
event_name: args.event_name,
user_identity: Identity { principal_id },
request_parameters: args.req_params,
@@ -345,11 +348,40 @@ impl Event {
}
}
fn serialize_event_time_millis<S>(value: &DateTime<Utc>, serializer: S) -> Result<S::Ok, S::Error>
fn serialize_event_time_millis<S>(value: &Timestamp, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
serializer.serialize_str(&value.to_rfc3339_opts(SecondsFormat::Millis, true))
serializer.serialize_str(&format_timestamp_rfc3339_millis(value))
}
fn deserialize_event_time_millis<'de, D>(deserializer: D) -> Result<Timestamp, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
value.parse::<Timestamp>().map_err(serde::de::Error::custom)
}
fn format_timestamp_rfc3339_millis(value: &Timestamp) -> String {
let utc = value.to_zoned(TimeZone::UTC);
format!(
"{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z",
utc.year(),
utc.month(),
utc.day(),
utc.hour(),
utc.minute(),
utc.second(),
utc.millisecond()
)
}
fn sequencer_from_timestamp(value: Timestamp) -> String {
match i64::try_from(value.as_nanosecond()) {
Ok(nanosecond) => format!("{nanosecond:X}"),
Err(_) => "0".to_string(),
}
}
fn initialize_response_elements(elements: &mut HashMap<String, String>, keys: &[&str]) {
@@ -550,6 +582,23 @@ mod tests {
assert_eq!(event.event_version, "2.3");
}
#[test]
fn event_new_preserves_legacy_sequencer_overflow_fallback() {
let args = EventArgsBuilder::new(
EventName::ObjectCreatedPut,
"bucket",
NotifyObjectInfo {
bucket: "bucket".to_string(),
name: "key".to_string(),
mod_time: Some(Timestamp::new(10_000_000_000, 0).expect("timestamp should be valid")),
..Default::default()
},
)
.build();
let event = Event::new(args);
assert_eq!(event.s3.object.sequencer, "0");
}
#[test]
fn object_restore_completed_includes_glacier_event_data() {
let args = EventArgsBuilder::new(
@@ -558,7 +607,7 @@ mod tests {
NotifyObjectInfo {
bucket: "bucket".to_string(),
name: "key".to_string(),
restore_expires: DateTime::<Utc>::from_timestamp(1_700_000_000, 0),
restore_expires: Timestamp::new(1_700_000_000, 0).ok(),
storage_class: Some("GLACIER".to_string()),
..Default::default()
},
@@ -696,7 +745,7 @@ mod tests {
#[test]
fn event_time_serializes_with_millisecond_precision() {
let mut event = Event::new_test_event("bucket", "key", EventName::ObjectCreatedPut);
event.event_time = DateTime::<Utc>::from_timestamp(1_711_423_698, 870_816_000).expect("timestamp should be valid");
event.event_time = Timestamp::new(1_711_423_698, 870_816_000).expect("timestamp should be valid");
let json = serde_json::to_value(&event).expect("event should serialize");
assert_eq!(json.get("eventTime").and_then(|value| value.as_str()), Some("2024-03-26T03:28:18.870Z"));
crates/policy/Cargo.toml
+1 -1
View File
@@ -66,7 +66,7 @@ base64-simd = { workspace = true }
jsonwebtoken = { workspace = true, features = ["aws_lc_rs"] }
regex = { workspace = true }
reqwest = { workspace = true, features = ["json"] }
chrono = { workspace = true, features = ["serde"] }
jiff = { workspace = true }
tracing.workspace = true
moka = { workspace = true, features = ["future"] }
async-trait.workspace = true
crates/policy/src/policy/opa.rs
+40 -1
View File
@@ -274,7 +274,9 @@ impl AuthZPlugin {
"context": {
"conditions": args.conditions,
"deny_only": args.deny_only,
"timestamp": chrono::Utc::now().to_rfc3339()
"timestamp": jiff::Timestamp::now()
.display_with_offset(jiff::tz::Offset::UTC)
.to_string()
}
}
})
@@ -439,6 +441,43 @@ mod tests {
}
}
#[test]
fn test_build_opa_input_timestamp_serializes_as_rfc3339_utc() {
let plugin = AuthZPlugin::new(Args {
url: "http://127.0.0.1:8181/v1/data/rustfs/authz/allow".to_string(),
auth_token: String::new(),
});
let groups = Some(vec!["developers".to_string()]);
let conditions = HashMap::new();
let claims = HashMap::new();
let args = PArgs {
account: "account",
groups: &groups,
action: crate::policy::action::Action::None,
bucket: "bucket",
conditions: &conditions,
is_owner: false,
object: "object",
claims: &claims,
deny_only: false,
};
let payload = plugin.build_opa_input(&args);
let timestamp = payload
.pointer("/input/context/timestamp")
.and_then(|value| value.as_str())
.expect("OPA input should include a timestamp string");
timestamp
.parse::<jiff::Timestamp>()
.expect("OPA timestamp should remain RFC3339-compatible");
assert!(timestamp.contains('T'), "OPA timestamp should use RFC3339 date-time form: {timestamp}");
assert!(
timestamp.ends_with("+00:00"),
"OPA timestamp should preserve chrono::DateTime::to_rfc3339 UTC offset form: {timestamp}"
);
}
#[test]
fn test_opa_connection_error_removes_sensitive_endpoint() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind local test listener");
crates/s3select-api/Cargo.toml
-1
View File
@@ -83,7 +83,6 @@ url.workspace = true
[dev-dependencies]
rustfs-test-utils.workspace = true
serial_test.workspace = true
tempfile.workspace = true
[lib]
doctest = false
crates/targets/Cargo.toml
+1 -1
View File
@@ -84,7 +84,7 @@ uuid = { workspace = true, features = ["v4", "v7", "serde", "fast-rng", "macro-d
sysinfo = { workspace = true, features = ["multithread"] }
rustfs-kafka-async = { workspace = true }
mysql_async = { workspace = true, default-features = false, features = ["default-rustls", "tracing"] }
chrono = { workspace = true, features = ["serde"] }
jiff = { workspace = true }
parking_lot = { workspace = true }
hashbrown = { workspace = true, features = ["serde", "rayon"] }
arc-swap = { workspace = true }
crates/targets/src/target/mysql.rs
+49 -3
View File
@@ -503,10 +503,35 @@ pub(crate) fn extract_event_time(body: &[u8]) -> Result<String, TargetError> {
.and_then(|v| v.as_str())
.ok_or_else(|| TargetError::Serialization("event_data is missing Records[0].eventTime".to_string()))?;
let dt = chrono::DateTime::parse_from_rfc3339(event_time)
let pieces = jiff::fmt::temporal::Pieces::parse(event_time)
.map_err(|e| TargetError::Serialization(format!("Failed to parse eventTime '{}': {}", event_time, e)))?;
let time = pieces
.time()
.ok_or_else(|| TargetError::Serialization(format!("Failed to parse eventTime '{}': missing RFC3339 time", event_time)))?;
if pieces.offset().is_none() {
return Err(TargetError::Serialization(format!(
"Failed to parse eventTime '{}': missing RFC3339 offset",
event_time
)));
}
if pieces.time_zone_annotation().is_some() {
return Err(TargetError::Serialization(format!(
"Failed to parse eventTime '{}': RFC3339 timestamp must not include a time zone annotation",
event_time
)));
}
let date = pieces.date();
Ok(dt.format("%Y-%m-%d %H:%M:%S%.6f").to_string())
Ok(format!(
"{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:06}",
date.year(),
date.month(),
date.day(),
time.hour(),
time.minute(),
time.second(),
time.subsec_nanosecond() / 1_000
))
}
/// Validates the required `event_time`/`event_data` columns and reports whether
@@ -1366,7 +1391,14 @@ mod tests {
let body =
br#"{"EventName":"s3:ObjectCreated:Put","Key":"bucket/obj.txt","Records":[{"eventTime":"2026-05-03T10:00:00Z"}]}"#;
let result = extract_event_time(body).expect("valid event_time");
assert!(result.starts_with("2026-05-03 10:00:00"));
assert_eq!(result, "2026-05-03 10:00:00.000000");
}
#[test]
fn extract_event_time_preserves_input_offset_wall_time() {
let body = br#"{"EventName":"s3:ObjectCreated:Put","Records":[{"eventTime":"2026-05-03T10:00:00.123456789+08:00"}]}"#;
let result = extract_event_time(body).expect("valid event_time");
assert_eq!(result, "2026-05-03 10:00:00.123456");
}
#[test]
@@ -1390,6 +1422,20 @@ mod tests {
assert!(err.to_string().contains("Failed to parse eventTime"));
}
#[test]
fn extract_event_time_without_offset_errors() {
let body = br#"{"Records":[{"eventTime":"2026-05-03T10:00:00"}]}"#;
let err = extract_event_time(body).expect_err("missing offset should fail");
assert!(err.to_string().contains("missing RFC3339 offset"));
}
#[test]
fn extract_event_time_with_time_zone_annotation_errors() {
let body = br#"{"Records":[{"eventTime":"2026-05-03T10:00:00+08:00[Asia/Shanghai]"}]}"#;
let err = extract_event_time(body).expect_err("time zone annotation should fail");
assert!(err.to_string().contains("must not include a time zone annotation"));
}
#[test]
fn extract_event_time_missing_records_errors() {
let body = br#"{"EventName":"s3:ObjectCreated:Put"}"#;
rustfs/src/server/event.rs
+11 -10
View File
@@ -21,7 +21,7 @@ use crate::storage_api::server::event::{
EventArgs as EcstoreEventArgs, StorageObjectInfo, read_existing_server_config_no_lock, register_event_dispatch_hook,
with_server_config_read_lock,
};
use chrono::{DateTime, Utc};
use jiff::Timestamp;
use rustfs_notify::{
EventArgs as NotifyEventArgs, NotificationError, NotificationRuntimeState, NotificationSystem, NotifyObjectInfo,
};
@@ -94,17 +94,18 @@ pub(crate) fn convert_ecstore_object_info(object: StorageObjectInfo) -> NotifyOb
.map(|(key, value)| (key.clone(), value.clone()))
.collect(),
version_id: object.version_id.map(|version_id| version_id.to_string()),
mod_time: object
.mod_time
.and_then(|value| DateTime::<Utc>::from_timestamp(value.unix_timestamp(), value.nanosecond())),
restore_expires: object
.restore_expires
.and_then(|value| DateTime::<Utc>::from_timestamp(value.unix_timestamp(), value.nanosecond())),
mod_time: object.mod_time.and_then(offset_date_time_to_timestamp),
restore_expires: object.restore_expires.and_then(offset_date_time_to_timestamp),
storage_class: object.storage_class,
transitioned_tier: (!object.transitioned_object.tier.is_empty()).then_some(object.transitioned_object.tier),
}
}
fn offset_date_time_to_timestamp(value: time::OffsetDateTime) -> Option<Timestamp> {
let nanosecond = value.nanosecond().try_into().ok()?;
Timestamp::new(value.unix_timestamp(), nanosecond).ok()
}
fn convert_ecstore_event_args(args: EcstoreEventArgs) -> Option<NotifyEventArgs> {
let version_id = args.object.version_id.map(|v| v.to_string()).unwrap_or_default();
let (host, port) = parse_host_and_port(args.host);
@@ -440,7 +441,7 @@ mod tests {
use crate::server::is_event_notifier_reconciled;
use crate::storage_api::server::event::StorageObjectInfo;
use crate::storage_api::server::event::contract::lifecycle::TransitionedObject;
use chrono::{DateTime, Utc};
use jiff::Timestamp;
use rustfs_notify::NotificationError;
use rustfs_notify::NotificationRuntimeState;
use serial_test::serial;
@@ -557,8 +558,8 @@ mod tests {
assert_eq!(converted.etag.as_deref(), Some("etag"));
assert_eq!(converted.content_type.as_deref(), Some("text/plain"));
assert_eq!(converted.user_defined.get("x-amz-meta-key").map(String::as_str), Some("value"));
assert_eq!(converted.mod_time, DateTime::<Utc>::from_timestamp(42, 0));
assert_eq!(converted.restore_expires, DateTime::<Utc>::from_timestamp(1_700_000_000, 0));
assert_eq!(converted.mod_time, Timestamp::new(42, 0).ok());
assert_eq!(converted.restore_expires, Timestamp::new(1_700_000_000, 0).ok());
assert_eq!(converted.storage_class.as_deref(), Some("GLACIER"));
assert_eq!(converted.transitioned_tier.as_deref(), Some("DEEP_ARCHIVE"));
}