diff --git a/docs/rfcs/RFC-007-library-event-stream.md b/docs/rfcs/RFC-007-library-event-stream.md index 22af20c..0c5f037 100644 --- a/docs/rfcs/RFC-007-library-event-stream.md +++ b/docs/rfcs/RFC-007-library-event-stream.md @@ -6,7 +6,9 @@ (l'appareil d'origine), [#159](https://github.com/InstaZDLL/waveflow-server/pull/159) (les événements d'album, décision 6), [#166](https://github.com/InstaZDLL/waveflow-server/pull/166) (la rétention, - décision 7). Le champ *Statut* ci-dessus ne bascule jamais dans ce + décision 7), + [#167](https://github.com/InstaZDLL/waveflow-server/pull/167) (l'acquittement, + décision 8). Le champ *Statut* ci-dessus ne bascule jamais dans ce projet : c'est cette ligne qui dit ce qui tourne, et elle se vérifie — une PR se lit, un mot de statut ne s'audite pas. - **Date** : 2026-08-25 @@ -293,7 +295,7 @@ sautant le trou. C'est exactement la panne que la décision 4 refuse. ## Décision 8 — l'acquittement, et ce qu'il ne décide pas -> **Décidée le 2026-08-30, pas encore construite.** +> **Construite.** Voir la ligne *Implémentée par* de l'en-tête. Une table `library_event_ack`, clé primaire `(library_id, device_id)`, portant le curseur et sa date. C'est `sync_ack` moins sa colonne de compte : là-bas la @@ -325,6 +327,6 @@ Plus rien de cette RFC n'est ouvert. Les trois questions qu'elle portait ont pour les deux. - ~~La forme exacte de l'acquittement.~~ Décision 8. -La décision 7 est construite ; la **décision 8 est décidée et pas encore -construite**. La ligne *Implémentée par* de l'en-tête ne nomme que ce qui -tourne, et c'est elle qu'il faut lire — pas cette section. +Les décisions 7 et 8 sont construites, et cette RFC n'a plus rien en attente. +La ligne *Implémentée par* de l'en-tête ne nomme que ce qui tourne, et c'est +elle qu'il faut lire — pas cette section. diff --git a/migrations-v2/20260830020000_library_event_ack.sql b/migrations-v2/20260830020000_library_event_ack.sql new file mode 100644 index 0000000..e61e62e --- /dev/null +++ b/migrations-v2/20260830020000_library_event_ack.sql @@ -0,0 +1,28 @@ +-- How far a device has read one library's feed. RFC-007 decision 8. +-- +-- `sync_ack` minus its account column, and the difference is the whole reason +-- this is a second table rather than a wider first one: the user journal is +-- keyed per account, this feed is keyed per library. A device belongs to +-- exactly one account (`device.user_id`), so an account column here would be a +-- third value derivable from the other two — a second truth to keep in +-- agreement with the first, and the sort that goes stale in one place only. +-- +-- The account is therefore re-read from the device and checked against +-- `library_member` at write time, which is the rule every other read in this +-- server follows: tenancy lives in the query. +-- +-- Both foreign keys cascade. A revoked device and a deleted library each leave +-- nothing behind, and neither needs a sweeper to notice. +-- +-- What this table deliberately does not do is hold back the purge. A device +-- that never comes back would pin a feed forever, and a shared library would +-- lose retention entirely the moment one phone was thrown away. Retention is +-- decided by RFC-007 decision 7 and reported against these rows; it is not +-- bounded by them. +CREATE TABLE library_event_ack ( + library_id TEXT NOT NULL REFERENCES library(id) ON DELETE CASCADE, + device_id TEXT NOT NULL REFERENCES device(id) ON DELETE CASCADE, + cursor INTEGER NOT NULL CHECK (cursor >= 0), + acknowledged_at INTEGER NOT NULL, + PRIMARY KEY (library_id, device_id) +) STRICT; diff --git a/src/api/libraries.rs b/src/api/libraries.rs index 0e74aa5..744b8eb 100644 --- a/src/api/libraries.rs +++ b/src/api/libraries.rs @@ -223,6 +223,55 @@ pub async fn scan_events( /// a different sequence and advances for different reasons, so it is a separate /// route with a separate cursor rather than a widening of the other — a rescan /// must not move a client's position in its own user journal. +/// How far a device has read one library's feed. +/// +/// A body rather than a header for the device, exactly like `/api/v2/sync/ack`: +/// the acknowledgement *is* about that device, so it is the request rather than +/// a note attached to it. The two are refused identically — 422 — for an +/// unknown or revoked device, a library the account cannot see, and a cursor +/// beyond what the feed has written. One answer for all three, because telling +/// them apart would say whether a library exists to somebody who may not know. +#[utoipa::path( + put, + path = "/api/v2/libraries/{library_id}/events/ack", + tag = "libraries", + params(("library_id" = Uuid, Path)), + request_body = LibraryEventAckRequest, + responses( + (status = 204), + (status = 401, body = ErrorResponse), + (status = 403, body = ErrorResponse), + (status = 422, body = ErrorResponse) + ) +)] +pub async fn library_events_ack( + State(state): State, + Path(library_id): Path, + headers: HeaderMap, + Json(request): Json, +) -> Result { + let user = authenticated(&state, &headers, Access::Write).await?; + let acknowledged = state + .services + .acknowledge_library_events(user.id, library_id, request.device_id, request.cursor) + .await + .map_err(service_error)?; + if !acknowledged { + return Err(ApiError::Validation); + } + Ok(StatusCode::NO_CONTENT) +} + +/// What a device says it has read. +#[derive(Debug, serde::Deserialize, utoipa::ToSchema)] +pub struct LibraryEventAckRequest { + pub device_id: Uuid, + /// The highest cursor this device has processed. Never lowered by the + /// server: a client that acknowledges an older cursor after a newer one has + /// raced its own two requests. + pub cursor: i64, +} + #[utoipa::path( get, path = "/api/v2/libraries/{library_id}/events", diff --git a/src/api/mod.rs b/src/api/mod.rs index 1d87200..2e230c2 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -103,6 +103,10 @@ pub fn router(state: AppState) -> Router { .route("/api/v2/scans/{scan_id}/events", get(scan_events)) .route("/api/v2/libraries/{library_id}/tracks", get(list_tracks)) .route("/api/v2/libraries/{library_id}/events", get(library_events)) + .route( + "/api/v2/libraries/{library_id}/events/ack", + put(library_events_ack), + ) // Its own body ceiling, and only its own. Raising the router's would // hand every route on the server a surface none of them asked for. .route( diff --git a/src/lib.rs b/src/lib.rs index 8f599f1..35f6cf7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -127,6 +127,7 @@ fn chunk_body_limit(limits: &config::UploadLimits) -> usize { api::scan_status, api::scan_events, api::library_events, + api::library_events_ack, api::negotiate_uploads, api::upload_session, api::upload_chunk, @@ -265,6 +266,7 @@ fn chunk_body_limit(limits: &config::UploadLimits) -> usize { sync::SyncPage, media::StreamTicketResponse, media::CanvasResponse, + api::LibraryEventAckRequest, scanner::ScanProgress )), modifiers(&SecurityAddon), diff --git a/src/services/library_events.rs b/src/services/library_events.rs index 8eba965..9a47922 100644 --- a/src/services/library_events.rs +++ b/src/services/library_events.rs @@ -20,6 +20,13 @@ pub struct EventPurge { pub events_removed: u64, /// Libraries that lost at least one. pub libraries_trimmed: usize, + /// Devices whose acknowledged cursor now sits below the watermark. + /// + /// They have been sent back to the catalogue snapshot. Counted rather than + /// prevented — RFC-007 decision 8 says the acknowledgement informs and does + /// not decide, or one forgotten phone would stop a shared library from ever + /// being trimmed. + pub devices_stranded: usize, } impl DomainServices { @@ -85,10 +92,18 @@ impl DomainServices { // writer gate is process-wide, and holding it across every feed on // a server with fifty libraries would stall every other mutation // for the whole pass. - let removed = self.purge_one_library(&library_id, cutoff).await?; + let (removed, stranded) = self.purge_one_library(&library_id, cutoff).await?; if removed > 0 { purged.events_removed += removed; purged.libraries_trimmed += 1; + if stranded > 0 { + tracing::info!( + library = %library_id, + devices = stranded, + "trimming this feed sent devices back to the catalogue" + ); + purged.devices_stranded += stranded; + } } } Ok(purged) @@ -101,7 +116,12 @@ impl DomainServices { /// window where the watermark claims less than has gone — and a client /// reading into it is handed a catch-up that looks complete while silently /// skipping the gap, which is the failure decision 4 exists to refuse. - async fn purge_one_library(&self, library_id: &str, cutoff: i64) -> Result { + /// Answers what was cut and how many devices the cut has *newly* stranded. + async fn purge_one_library( + &self, + library_id: &str, + cutoff: i64, + ) -> Result<(u64, usize), ServiceError> { let _writer = self.db.writer_guard().await; let mut tx = self.db.pool().begin().await?; @@ -129,7 +149,7 @@ impl DomainServices { let removed: i64 = eligible.try_get("n")?; let highest: Option = eligible.try_get("highest")?; let (Some(highest), true) = (highest, removed > 0) else { - return Ok(0); + return Ok((0, 0)); }; sqlx::query( @@ -145,6 +165,29 @@ impl DomainServices { .execute(&mut *tx) .await?; + // Counted between the two watermarks rather than below the new one: + // this pass is answering for what *it* cost. A device already below the + // old watermark was sent back to the catalogue by an earlier pass, and + // counting it again every time the feed is trimmed would report a bill + // that only ever grows. + // + // Inside the transaction because both watermarks are known here and + // nowhere else — read afterwards, the old one is already gone. + let previous: i64 = + sqlx::query_scalar("SELECT events_purged_through FROM library WHERE id=?") + .bind(library_id) + .fetch_one(&mut *tx) + .await?; + let stranded: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM library_event_ack \ + WHERE library_id=? AND cursor >= ? AND cursor < ?", + ) + .bind(library_id) + .bind(previous) + .bind(highest) + .fetch_one(&mut *tx) + .await?; + // `MAX` never decreases: a pass that cut an older tail must not lower a // watermark an earlier one raised, and two passes racing must not // either. @@ -156,7 +199,111 @@ impl DomainServices { .execute(&mut *tx) .await?; tx.commit().await?; - Ok(u64::try_from(removed).unwrap_or(0)) + Ok(( + u64::try_from(removed).unwrap_or(0), + usize::try_from(stranded).unwrap_or(0), + )) + } + + /// Records how far a device has read one library's feed. + /// + /// RFC-007 decision 8. Two checks rather than one, and both are in the + /// statement: the device must be this account's and unrevoked, and the + /// account must be a member of the library. `sync_ack` needs only the + /// first, because the journal is keyed per account and there is no second + /// scope to escape into; here there is. + /// + /// `false` for anything refused — an unknown or revoked device, a library + /// this account cannot see, a cursor beyond what the feed has written. A + /// caller who is not a member learns nothing about whether the library + /// exists, which is the rule everywhere else in this API. + pub async fn acknowledge_library_events( + &self, + user_id: Uuid, + library_id: Uuid, + device_id: Uuid, + cursor: i64, + ) -> Result { + if cursor < 0 { + return Ok(false); + } + let _writer = self.db.writer_guard().await; + let mut tx = self.db.pool().begin().await?; + + // A cursor beyond what the feed has written would let a client mark + // itself caught up with events that do not exist yet, and then be + // silently behind when they arrive. + // + // Scoped by membership as well, and that is the project's rule rather + // than a leak being closed: today both paths out of this function + // answer 422, so a caller cannot tell "beyond the feed" from "not your + // library" whichever way this reads. But tenancy lives in the query + // here, never in a check the query trusts — `library_changes` keeps a + // redundant membership predicate for the same reason, so that removing + // one guard cannot quietly widen another. + // The maximum is a *subquery* rather than an aggregate over the join, + // and that is not style. `SELECT MAX(...) FROM ... WHERE ` + // returns one row holding NULL, so `COALESCE(..., 0)` would hand back + // `Some(0)` for a library the caller cannot see and the predicate above + // would decide nothing. Non-aggregate over `library_member`, zero + // matching rows means zero rows returned, which is the answer wanted. + // The watermark counts as well as the surviving rows, because the + // question is "the furthest position a client could legitimately have + // reached" and a purge does not move that backwards. Read from the rows + // alone, a feed trimmed to nothing would answer 0 and refuse an + // acknowledgement at a cursor `library_changes` accepts to read from — + // the two would disagree about the same number. + // + // Unreachable while `min_events` must be positive, since a floor of one + // leaves a row whose cursor is at or above the watermark. That is an + // invariant enforced three files away in `parse_positive_env`, and this + // is the expression that does not depend on it. + let latest: Option = sqlx::query_scalar( + "SELECT MAX( \ + l.events_purged_through, \ + COALESCE((SELECT MAX(e.cursor) FROM library_event e \ + WHERE e.library_id=m.library_id), 0)) \ + FROM library_member m \ + JOIN device d ON d.user_id=m.user_id \ + JOIN library l ON l.id=m.library_id \ + WHERE m.library_id=? AND m.user_id=? AND d.id=? AND d.revoked_at IS NULL", + ) + .bind(library_id.to_string()) + .bind(user_id.to_string()) + .bind(device_id.to_string()) + .fetch_optional(&mut *tx) + .await?; + let Some(latest) = latest else { + return Ok(false); + }; + if cursor > latest { + return Ok(false); + } + + let result = sqlx::query( + "INSERT INTO library_event_ack (library_id, device_id, cursor, acknowledged_at) \ + SELECT ?, ?, ?, ? WHERE EXISTS ( \ + SELECT 1 FROM device d \ + JOIN library_member m ON m.user_id=d.user_id \ + WHERE d.id=? AND d.user_id=? AND d.revoked_at IS NULL AND m.library_id=? \ + ) ON CONFLICT (library_id, device_id) DO UPDATE SET \ + cursor=MAX(library_event_ack.cursor, excluded.cursor), \ + acknowledged_at=excluded.acknowledged_at", + ) + .bind(library_id.to_string()) + .bind(device_id.to_string()) + .bind(cursor) + .bind(now_ms()) + .bind(device_id.to_string()) + .bind(user_id.to_string()) + .bind(library_id.to_string()) + .execute(&mut *tx) + .await?; + tx.commit().await?; + // Never lowered: a client that acknowledges an older cursor after a + // newer one has raced its own two requests, and the server is not the + // place to decide which of them is the truth. + Ok(result.rows_affected() == 1) } /// One page of a library's changes, for a caller entitled to that library. diff --git a/tests/catalog.rs b/tests/catalog.rs index 9a0caeb..0a373e4 100644 --- a/tests/catalog.rs +++ b/tests/catalog.rs @@ -1919,3 +1919,304 @@ async fn retention_cuts_by_age_never_below_the_floor_and_never_at_the_bound() { "the floor kept a usable tail rather than an empty feed" ); } + +/// RFC-007 decision 8. The acknowledgement says how far a device has read; what +/// it deliberately does not do is hold the purge back. +#[tokio::test] +async fn an_acknowledgement_records_a_device_without_holding_the_purge_back() { + let temp = tempfile::tempdir().unwrap(); + let mut config = waveflow_server::Config::for_data_dir(temp.path().join("data")); + config.library_event_retention.min_events = 1; + config.library_event_retention.days = 30; + let state = waveflow_server::initialize(&config).await.unwrap(); + + let password = security::generate_token("test-password-"); + let hash = security::hash_password(&password).unwrap(); + let owner = state + .db + .create_account("ack", &hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let music = config.data_dir.join("ack-music"); + std::fs::create_dir_all(&music).unwrap(); + let library = state + .db + .create_library( + owner, + "Ack", + &std::fs::canonicalize(&music).unwrap(), + LibraryVisibility::Private, + now_ms(), + ) + .await + .unwrap(); + + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 3, false).await.unwrap(); + for index in 0..3usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + + let router = waveflow_server::app(&config, state.clone()); + let (token, device) = login_session(&router, "ack", &password).await; + let cursors: Vec = + sqlx::query_scalar("SELECT cursor FROM library_event WHERE library_id=? ORDER BY cursor") + .bind(library.to_string()) + .fetch_all(state.db.pool()) + .await + .unwrap(); + + let ack = |cursor: i64, device: String, token: String| { + let router = router.clone(); + async move { + router + .oneshot( + Request::put(format!("/api/v2/libraries/{library}/events/ack")) + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ "device_id": device, "cursor": cursor }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap() + .status() + } + }; + + assert_eq!( + ack(cursors[0], device.clone(), token.clone()).await, + StatusCode::NO_CONTENT + ); + // A cursor beyond what the feed has written would let a client mark itself + // caught up with events that do not exist yet, and be silently behind when + // they arrive. + assert_eq!( + ack(cursors[2] + 1, device.clone(), token.clone()).await, + StatusCode::UNPROCESSABLE_ENTITY + ); + // Somebody else's device, refused the same way a library one cannot see is + // — one answer for all of them, or the refusal tells them apart. + let stranger_hash = security::hash_password("correct horse battery staple").unwrap(); + state + .db + .create_account("ack-stranger", &stranger_hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let (stranger_token, stranger_device) = + login_session(&router, "ack-stranger", "correct horse battery staple").await; + assert_eq!( + ack(cursors[0], stranger_device.clone(), token.clone()).await, + StatusCode::UNPROCESSABLE_ENTITY, + "a device that is not this account's" + ); + // Their own device, and a library they are not a member of. This is the + // case the membership join answers — the one above is refused a step + // earlier, by the device check, so it says nothing about tenancy. + assert_eq!( + ack(cursors[0], stranger_device, stranger_token).await, + StatusCode::UNPROCESSABLE_ENTITY, + "a library this account is not a member of" + ); + + // Never lowered: two of a client's own requests racing must not let the + // older one win. + assert_eq!( + ack(cursors[2], device.clone(), token.clone()).await, + StatusCode::NO_CONTENT + ); + assert_eq!( + ack(cursors[0], device.clone(), token).await, + StatusCode::NO_CONTENT + ); + let stored: i64 = sqlx::query_scalar( + "SELECT cursor FROM library_event_ack WHERE library_id=? AND device_id=?", + ) + .bind(library.to_string()) + .bind(&device) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert_eq!(stored, cursors[2], "the highest acknowledged cursor stands"); + + // And the point of the whole table: it does not hold the purge back. This + // device has acknowledged everything, then the feed is aged out anyway — + // one forgotten phone must not stop a shared library being trimmed. + sqlx::query("UPDATE library_event_ack SET cursor=? WHERE library_id=?") + .bind(cursors[0]) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let now = now_ms(); + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * 24 * 60 * 60 * 1000i64) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let purged = state.services.purge_library_events(now).await.unwrap(); + assert_eq!( + purged.events_removed, 2, + "the floor of one keeps the newest" + ); + assert_eq!( + purged.devices_stranded, 1, + "and the cut is reported against the device it sent back" + ); + + // A pass answers for what it cost, not for what it inherits. Write two more + // events, age them, and trim again: the device is still below the + // watermark, but this pass is not what put it there. + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 2, false).await.unwrap(); + for index in 3..5usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * 24 * 60 * 60 * 1000i64) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let purged = state.services.purge_library_events(now).await.unwrap(); + assert!(purged.events_removed > 0, "there was something left to cut"); + assert_eq!( + purged.devices_stranded, 0, + "already stranded is not stranded again — a bill that only grows is not a bill" + ); +} + +/// The acknowledgement's ceiling counts the watermark, not only the rows. +/// +/// **The configuration here is one the server would refuse to start with.** +/// `parse_positive_env` rejects a floor of zero, so a feed can never actually +/// be trimmed to nothing in production, and this state is unreachable. The +/// guard exists anyway because "the furthest position a client could have +/// reached" should not be an expression that happens to be right thanks to a +/// bound enforced three files away — and a test that never reaches the branch +/// is a guard nobody can check. `Config::for_data_dir` does not validate, which +/// is what lets this be written at all. +#[tokio::test] +async fn a_cursor_at_the_watermark_is_acknowledged_even_with_the_feed_emptied() { + let temp = tempfile::tempdir().unwrap(); + let mut config = waveflow_server::Config::for_data_dir(temp.path().join("data")); + config.library_event_retention.min_events = 0; + config.library_event_retention.days = 30; + let state = waveflow_server::initialize(&config).await.unwrap(); + + let password = security::generate_token("test-password-"); + let hash = security::hash_password(&password).unwrap(); + let owner = state + .db + .create_account("watermark-ack", &hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let music = config.data_dir.join("watermark-ack-music"); + std::fs::create_dir_all(&music).unwrap(); + let library = state + .db + .create_library( + owner, + "Watermark", + &std::fs::canonicalize(&music).unwrap(), + LibraryVisibility::Private, + now_ms(), + ) + .await + .unwrap(); + + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 2, false).await.unwrap(); + for index in 0..2usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + + let now = now_ms(); + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * 24 * 60 * 60 * 1000i64) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + state.services.purge_library_events(now).await.unwrap(); + + let remaining: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM library_event WHERE library_id=?") + .bind(library.to_string()) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert_eq!(remaining, 0, "the floor of zero let the feed empty"); + let watermark: i64 = sqlx::query_scalar("SELECT events_purged_through FROM library WHERE id=?") + .bind(library.to_string()) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert!(watermark > 0); + + // `library_changes` accepts reading from the watermark, so the + // acknowledgement has to accept recording it — the two must not disagree + // about the same number. + let router = waveflow_server::app(&config, state.clone()); + let (token, device) = login_session(&router, "watermark-ack", &password).await; + assert!(state + .services + .library_changes(owner, library, watermark, 500) + .await + .is_ok()); + let response = router + .oneshot( + Request::put(format!("/api/v2/libraries/{library}/events/ack")) + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ "device_id": device, "cursor": watermark }).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NO_CONTENT); +}