relay: skip TTL deadline bump for known-permanent channels (T1a write-amp) (#2125)

Signed-off-by: Tyler Longwell <tlongwell@block.xyz> Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Co-authored-by: Tyler Longwell <tlongwell@block.xyz> Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
2026-08-18 06:50:31 +02:00 · 2026-07-19 14:55:51 -04:00
parent d04a4c978a
commit 2e936d439c
8 changed files with 403 additions and 42 deletions
@@ -0,0 +1,77 @@
+#!/usr/bin/env bash
+# T1a correctness evidence (runnable from this lane tip).
+#
+# Proves, against a live relay built from this tree:
+#   1. permanent channel: kind:9 ingest emits no separate top-level TTL UPDATE
+#      transaction (the deferred trigger's conditional statement stays inside the
+#      event transaction and affects zero rows);
+#   2. ephemeral channel: the TTL bump is still observed (ttl_deadline strictly advances
+#      across a message);
+#   3. TTL-set-during-ingest race: messages committed after concurrent TTL activation
+#      extend the deadline beyond the activation update's own deadline.
+#
+# Usage: bench/t1a_evidence.sh <pg_container> <db_url> <relay_ws_url> <community_uuid>
+# Requires: relay running FROM THIS TREE against <db_url>; psql via docker exec;
+#           BENCH_PRIVATE_KEY env (member secret key hex); wamp_bench built --release.
+set -euo pipefail
+
+PG="$1"; DBURL="$2"; RELAY="$3"; COMMUNITY="$4"
+PSQL=(docker exec "$PG" psql "$DBURL" -tA)
+sql() { "${PSQL[@]}" -c "$1"; }
+BIN="./target/release/wamp_bench"
+# BENCH_PUB = x-only pubkey hex for BENCH_PRIVATE_KEY (both required).
+PUB="${BENCH_PUB:?set BENCH_PUB (x-only pubkey hex matching BENCH_PRIVATE_KEY)}"
+
+mkchan() { # $1 name, $2 ttl_seconds or NULL -> echoes uuid
+  local id; id=$(python3 -c "import uuid;print(uuid.uuid4())")
+  sql "insert into channels (id, community_id, name, channel_type, visibility, created_by, ttl_seconds, ttl_deadline)
+       values ('$id','$COMMUNITY','$1','stream','private',decode('$PUB','hex'),$2,
+               case when $2::int is null then null else now() + ($2::int || ' seconds')::interval end)" >/dev/null
+  sql "insert into channel_members (community_id, channel_id, pubkey, role)
+       values ('$COMMUNITY','$id',decode('$PUB','hex'),'owner')" >/dev/null
+  echo "$id"
+}
+
+FAIL=0
+
+echo "== 1. permanent channel: zero separate top-level TTL UPDATE statements =="
+PERM=$(mkchan t1a-perm NULL)
+sql "select pg_stat_statements_reset()" >/dev/null
+env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$PERM" 20 10 2 /tmp/t1a-perm.lat >/tmp/t1a-perm.json
+ACC=$(python3 -c "import json;print(json.load(open('/tmp/t1a-perm.json'))['accepted'])")
+TTL_CALLS=$(sql "select coalesce(sum(calls),0) from pg_stat_statements where query ilike '%UPDATE channels SET ttl_deadline%'")
+echo "accepted=$ACC ttl_update_calls=$TTL_CALLS"
+[[ "$ACC" -gt 0 && "$TTL_CALLS" == "0" ]] || { echo "FAIL: expected >0 accepted and 0 TTL updates"; FAIL=1; }
+
+echo "== 2. ephemeral channel: bump still observed =="
+EPH=$(mkchan t1a-eph 3600)
+D0=$(sql "select extract(epoch from ttl_deadline) from channels where id='$EPH'")
+sleep 2
+env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$EPH" 5 3 1 /tmp/t1a-eph.lat >/tmp/t1a-eph.json
+D1=$(sql "select extract(epoch from ttl_deadline) from channels where id='$EPH'")
+echo "deadline before=$D0 after=$D1"
+python3 -c "import sys; sys.exit(0 if float('$D1') > float('$D0') else 1)" \
+  || { echo "FAIL: ephemeral ttl_deadline did not advance"; FAIL=1; }
+
+echo "== 3. TTL-set-during-ingest race =="
+RACE=$(mkchan t1a-race NULL)
+env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$RACE" 50 6 4 /tmp/t1a-race.lat >/tmp/t1a-race.json &
+BPID=$!
+sleep 2
+# update_channel-equivalent: set TTL and reset deadline in one statement, mid-burst.
+ACTIVATION_DEADLINE=$(sql "update channels set ttl_seconds=600, ttl_deadline=clock_timestamp() + interval '600 seconds', updated_at=now() where id='$RACE' and deleted_at is null returning extract(epoch from ttl_deadline)")
+wait "$BPID"
+ROW=$(sql "select ttl_seconds, extract(epoch from ttl_deadline) from channels where id='$RACE'")
+echo "activation_deadline=$ACTIVATION_DEADLINE post-race=$ROW"
+FINAL_DEADLINE="${ROW#*|}"
+python3 -c "import sys; sys.exit(0 if float('$FINAL_DEADLINE') > float('$ACTIVATION_DEADLINE') else 1)" \
+  || { echo "FAIL: later message did not extend TTL beyond activation deadline"; FAIL=1; }
+# and subsequent messages now bump it (channel is ephemeral now)
+D0=$(sql "select extract(epoch from ttl_deadline) from channels where id='$RACE'")
+sleep 2
+env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$RACE" 5 3 1 /tmp/t1a-race2.lat >/tmp/t1a-race2.json
+D1=$(sql "select extract(epoch from ttl_deadline) from channels where id='$RACE'")
+python3 -c "import sys; sys.exit(0 if float('$D1') > float('$D0') else 1)" \
+  || { echo "FAIL: post-race ephemeral bump not observed"; FAIL=1; }
+
+[[ "$FAIL" == 0 ]] && echo "T1A EVIDENCE: ALL PASS" || { echo "T1A EVIDENCE: FAILURES"; exit 1; }
@@ -1312,25 +1312,6 @@ pub async fn get_member_role(
    Ok(row.map(|r| r.try_get("role")).transpose()?)
 }

-/// Bump the TTL deadline for an ephemeral channel after a new message.
-///
-/// No-op for permanent channels or channels that are already archived/deleted.
-pub async fn bump_ttl_deadline(
-    pool: &PgPool,
-    community_id: CommunityId,
-    channel_id: Uuid,
-) -> Result<()> {
-    sqlx::query(
-        "UPDATE channels SET ttl_deadline = NOW() + (ttl_seconds || ' seconds')::interval \
-         WHERE community_id = $1 AND id = $2 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL",
-    )
-    .bind(community_id.as_uuid())
-    .bind(channel_id)
-    .execute(pool)
-    .await?;
-    Ok(())
-}
-
 /// Archive ephemeral channels whose TTL deadline has passed.
 ///
 /// Returns the `(community_id, host, channel_id)` list that was archived. Idempotent — the
@@ -1454,6 +1454,157 @@ mod tests {
        id
    }

+    async fn make_test_channel(
+        pool: &PgPool,
+        community_id: Uuid,
+        ttl_seconds: Option<i32>,
+    ) -> Uuid {
+        let id = Uuid::new_v4();
+        sqlx::query(
+            "INSERT INTO channels \
+             (id, community_id, name, created_by, ttl_seconds, ttl_deadline) \
+             VALUES ($1, $2, $3, $4, $5, \
+                     CASE WHEN $5 IS NULL THEN NULL \
+                          ELSE clock_timestamp() + make_interval(secs => $5) END)",
+        )
+        .bind(id)
+        .bind(community_id)
+        .bind(format!("event-ttl-test-{}", id.simple()))
+        .bind(vec![7_u8; 32])
+        .bind(ttl_seconds)
+        .execute(pool)
+        .await
+        .expect("insert test channel");
+        id
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn event_insert_ttl_trigger_handles_permanent_ephemeral_duplicate_and_activation_race() {
+        let pool = setup_pool().await;
+        let community_uuid = make_test_community(&pool).await;
+        let community = CommunityId::from_uuid(community_uuid);
+
+        let permanent = make_test_channel(&pool, community_uuid, None).await;
+        let permanent_event = make_text_event("permanent channel event");
+        assert!(
+            insert_event(&pool, community, &permanent_event, Some(permanent))
+                .await
+                .expect("insert permanent event")
+                .1
+        );
+        let permanent_deadline: Option<DateTime<Utc>> = sqlx::query_scalar(
+            "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(permanent)
+        .fetch_one(&pool)
+        .await
+        .expect("read permanent deadline");
+        assert_eq!(permanent_deadline, None);
+
+        let ephemeral = make_test_channel(&pool, community_uuid, Some(60)).await;
+        let initial_deadline: DateTime<Utc> = sqlx::query_scalar(
+            "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(ephemeral)
+        .fetch_one(&pool)
+        .await
+        .expect("read initial ephemeral deadline");
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+        let ephemeral_event = make_text_event("ephemeral channel event");
+        assert!(
+            insert_event(&pool, community, &ephemeral_event, Some(ephemeral))
+                .await
+                .expect("insert ephemeral event")
+                .1
+        );
+        let bumped_deadline: DateTime<Utc> = sqlx::query_scalar(
+            "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(ephemeral)
+        .fetch_one(&pool)
+        .await
+        .expect("read bumped ephemeral deadline");
+        assert!(bumped_deadline > initial_deadline);
+        assert!(
+            !insert_event(&pool, community, &ephemeral_event, Some(ephemeral))
+                .await
+                .expect("insert duplicate event")
+                .1
+        );
+        let duplicate_deadline: DateTime<Utc> = sqlx::query_scalar(
+            "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(ephemeral)
+        .fetch_one(&pool)
+        .await
+        .expect("read deadline after duplicate");
+        assert_eq!(duplicate_deadline, bumped_deadline);
+
+        // Reproduce the blocked stale-prefetch ordering: ingest has already
+        // observed a permanent channel, then TTL activation locks/updates the
+        // row before the event INSERT reaches its trigger. The trigger must
+        // wait and refresh from the later event after activation commits.
+        let racing = make_test_channel(&pool, community_uuid, None).await;
+        let stale_ttl: Option<i32> = sqlx::query_scalar(
+            "SELECT ttl_seconds FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(racing)
+        .fetch_one(&pool)
+        .await
+        .expect("prefetch permanent channel");
+        assert_eq!(stale_ttl, None);
+
+        let mut activation = pool.begin().await.expect("begin TTL activation");
+        let activation_deadline: DateTime<Utc> = sqlx::query_scalar(
+            "UPDATE channels \
+             SET ttl_seconds = 60, ttl_deadline = clock_timestamp() + interval '60 seconds' \
+             WHERE community_id = $1 AND id = $2 RETURNING ttl_deadline",
+        )
+        .bind(community_uuid)
+        .bind(racing)
+        .fetch_one(&mut *activation)
+        .await
+        .expect("activate TTL while holding channel row lock");
+
+        let race_pool = pool.clone();
+        let racing_event = make_text_event("event after stale permanent prefetch");
+        let insert = tokio::spawn(async move {
+            insert_event(&race_pool, community, &racing_event, Some(racing)).await
+        });
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        assert!(
+            !insert.is_finished(),
+            "event trigger must wait on TTL activation"
+        );
+        activation.commit().await.expect("commit TTL activation");
+        assert!(
+            insert
+                .await
+                .expect("join racing insert")
+                .expect("racing insert")
+                .1
+        );
+
+        let final_deadline: DateTime<Utc> = sqlx::query_scalar(
+            "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(racing)
+        .fetch_one(&pool)
+        .await
+        .expect("read deadline after racing event");
+        assert!(
+            final_deadline > activation_deadline + chrono::Duration::milliseconds(50),
+            "later event must extend TTL beyond activation deadline: activation={activation_deadline}, final={final_deadline}"
+        );
+    }
+
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_event_by_id_is_scoped_when_event_id_collides_across_communities() {
@@ -1688,15 +1688,6 @@ impl Db {
        channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await
    }

-    /// Bump the TTL deadline for an ephemeral channel after a new message.
-    pub async fn bump_ttl_deadline(
-        &self,
-        community_id: CommunityId,
-        channel_id: Uuid,
-    ) -> Result<()> {
-        channel::bump_ttl_deadline(&self.pool, community_id, channel_id).await
-    }
-
    /// Archive ephemeral channels whose TTL deadline has passed.
    pub async fn reap_expired_ephemeral_channels(
        &self,
@@ -532,9 +532,12 @@ mod tests {
                let normalized = normalize_sql(statement);
                let updates_channels =
                    identifier_after_keyword(statement, "update").as_deref() == Some("channels");
+                let update_assignments = normalized
+                    .split_once(" set ")
+                    .map(|(_, tail)| tail.split_once(" where ").map_or(tail, |(set, _)| set));
                let mutates_with_update = updates_channels
-                    && normalized.contains(" set ")
-                    && normalized.contains("community_id");
+                    && update_assignments
+                        .is_some_and(|assignments| assignments.contains("community_id"));
                let alters_channels = identifier_after_keyword(statement, "alter table").as_deref()
                    == Some("channels");
                let drops_channels = identifier_after_keyword(statement, "drop table").as_deref()
@@ -557,7 +560,7 @@ mod tests {
        let mut migrations: Vec<_> = MIGRATOR.iter().collect();
        migrations.sort_by_key(|migration| migration.version);

-        assert_eq!(migrations.len(), 21);
+        assert_eq!(migrations.len(), 22);
        assert_eq!(migrations[0].version, 1);
        assert_eq!(&*migrations[0].description, "initial schema");
        assert!(migrations[0]
@@ -839,6 +842,16 @@ mod tests {
            .sql
            .as_str()
            .contains("join_policy_acceptances"));
+
+        // Channel TTL refresh belongs to the event insertion transaction so a
+        // concurrent permanent -> ephemeral transition cannot be missed.
+        assert_eq!(migrations[21].version, 22);
+        let ttl_refresh = migrations[21].sql.as_str();
+        assert!(ttl_refresh.contains("CREATE CONSTRAINT TRIGGER events_refresh_channel_ttl"));
+        assert!(ttl_refresh.contains("AFTER INSERT ON events"));
+        assert!(ttl_refresh.contains("DEFERRABLE INITIALLY DEFERRED"));
+        assert!(ttl_refresh.contains("clock_timestamp()"));
+        assert!(ttl_refresh.contains("NEW.kind <> 9007"));
    }

    #[test]
@@ -1081,7 +1094,7 @@ mod tests {
        run_migrations(&pool)
            .await
            .expect("retry succeeds after operator repair");
-        assert_eq!(applied_versions(&pool).await.last().copied(), Some(21));
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(22));
    }

    #[tokio::test]
@@ -2351,16 +2351,6 @@ async fn ingest_event_inner(
        });
    }

-    // Any successfully stored channel-scoped event keeps the channel alive.
-    // Skip kind:9007 (create) — the deadline was just set during creation.
-    if let Some(ch_id) = channel_id {
-        if kind_u32 != KIND_NIP29_CREATE_GROUP {
-            if let Err(e) = state.db.bump_ttl_deadline(tenant.community(), ch_id).await {
-                warn!(channel = %ch_id, "TTL deadline bump failed: {e}");
-            }
-        }
-    }
-
    if crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
        if let Err(e) =
            crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state)
@@ -0,0 +1,118 @@
+//! Paced kind:9 load generator for relay write-amplification benchmarking.
+//!
+//! Opens `conns` authenticated WebSocket connections (one shared identity)
+//! and sends kind:9 text events to `channel_uuid` at a total rate of `qps`
+//! for `duration_secs`. Each connection is synchronous (send -> await OK),
+//! paced by a per-connection tokio interval, so OK latency is measured
+//! end-to-end. Emits a JSON summary on stdout and one raw latency sample
+//! (milliseconds, f64) per line to `latency_out`.
+//!
+//! Usage: wamp-bench <channel_uuid> <qps> <duration_secs> <conns> <latency_out>
+//! Env:   BUZZ_RELAY_URL (default ws://localhost:3000), BENCH_PRIVATE_KEY (hex)
+
+use std::time::{Duration, Instant};
+
+use buzz_test_client::BuzzTestClient;
+use nostr::Keys;
+use tokio::time::MissedTickBehavior;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let _ = rustls::crypto::CryptoProvider::install_default(
+        rustls::crypto::aws_lc_rs::default_provider(),
+    );
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() != 6 {
+        eprintln!("Usage: wamp-bench <channel_uuid> <qps> <duration_secs> <conns> <latency_out>");
+        std::process::exit(1);
+    }
+    let channel_id = args[1].clone();
+    let qps: f64 = args[2].parse()?;
+    let duration_secs: u64 = args[3].parse()?;
+    let conns: usize = args[4].parse()?;
+    let latency_out = args[5].clone();
+
+    let url = std::env::var("BUZZ_RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".into());
+    let keys = match std::env::var("BENCH_PRIVATE_KEY") {
+        Ok(hex) => Keys::parse(&hex)?,
+        Err(_) => anyhow::bail!("BENCH_PRIVATE_KEY is required (channel member secret key)"),
+    };
+
+    let per_conn_interval = Duration::from_secs_f64(conns as f64 / qps);
+    let deadline = Instant::now() + Duration::from_secs(duration_secs);
+
+    let mut tasks = Vec::new();
+    for conn_idx in 0..conns {
+        let url = url.clone();
+        let keys = keys.clone();
+        let channel_id = channel_id.clone();
+        tasks.push(tokio::spawn(async move {
+            let mut client = BuzzTestClient::connect(&url, &keys).await?;
+            let mut interval = tokio::time::interval(per_conn_interval);
+            interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
+            let mut latencies: Vec<f64> = Vec::new();
+            let mut sent: u64 = 0;
+            let mut rejected: u64 = 0;
+            let mut seq: u64 = 0;
+            while Instant::now() < deadline {
+                interval.tick().await;
+                seq += 1;
+                let content = format!(
+                    "wamp-bench c{conn_idx} m{seq} payload: the quick brown fox jumps over the lazy dog 0123456789"
+                );
+                let start = Instant::now();
+                let ok = client
+                    .send_text_message(&keys, &channel_id, &content, 9)
+                    .await?;
+                let elapsed_ms = start.elapsed().as_secs_f64() * 1e3;
+                sent += 1;
+                if ok.accepted {
+                    latencies.push(elapsed_ms);
+                } else {
+                    rejected += 1;
+                    eprintln!("REJECTED conn={conn_idx} seq={seq}: {}", ok.message);
+                }
+            }
+            client.disconnect().await?;
+            Ok::<_, anyhow::Error>((sent, rejected, latencies))
+        }));
+    }
+
+    let mut sent = 0u64;
+    let mut rejected = 0u64;
+    let mut latencies: Vec<f64> = Vec::new();
+    for task in tasks {
+        let (s, r, l) = task.await??;
+        sent += s;
+        rejected += r;
+        latencies.extend(l);
+    }
+    latencies.sort_by(|a, b| a.partial_cmp(b).expect("finite latencies"));
+    let pct = |p: f64| -> f64 {
+        if latencies.is_empty() {
+            return f64::NAN;
+        }
+        let idx = ((latencies.len() as f64 - 1.0) * p).round() as usize;
+        latencies[idx]
+    };
+    let raw: String = latencies.iter().map(|l| format!("{l:.3}\n")).collect();
+    std::fs::write(&latency_out, raw)?;
+    println!(
+        "{}",
+        serde_json::json!({
+            "sent": sent,
+            "accepted": sent - rejected,
+            "rejected": rejected,
+            "qps_target": qps,
+            "duration_secs": duration_secs,
+            "conns": conns,
+            "ok_latency_ms": {
+                "p50": pct(0.50),
+                "p95": pct(0.95),
+                "p99": pct(0.99),
+                "max": latencies.last().copied().unwrap_or(f64::NAN),
+            },
+        })
+    );
+    Ok(())
+}
@@ -0,0 +1,40 @@
+-- Refresh ephemeral-channel expiry in the transaction that makes a
+-- channel-scoped event durable. Deferring to COMMIT closes the stale-prefetch
+-- race: the UPDATE sees a TTL transition committed while ingest was in flight,
+-- or waits on its row lock and rechecks after it commits, without restoring a
+-- separate hot-path transaction.
+CREATE FUNCTION refresh_channel_ttl_after_event_insert() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    -- Kind 9007 creates the channel and initializes its deadline itself.
+    IF NEW.channel_id IS NOT NULL AND NEW.kind <> 9007 THEN
+        BEGIN
+            -- Lock by identity before testing ttl_seconds. If a concurrent TTL
+            -- transition is uncommitted, this waits and follows its updated row
+            -- version instead of treating the old permanent version as final.
+            PERFORM 1 FROM channels
+            WHERE community_id = NEW.community_id AND id = NEW.channel_id
+            FOR UPDATE;
+
+            UPDATE channels
+            SET ttl_deadline = clock_timestamp() + make_interval(secs => ttl_seconds)
+            WHERE community_id = NEW.community_id
+              AND id = NEW.channel_id
+              AND ttl_seconds IS NOT NULL
+              AND archived_at IS NULL
+              AND deleted_at IS NULL;
+        EXCEPTION WHEN OTHERS THEN
+            -- Preserve the existing best-effort contract: a TTL refresh failure
+            -- must not reject an otherwise valid durable event.
+            RAISE WARNING 'channel TTL refresh failed for community %, channel %: %',
+                NEW.community_id, NEW.channel_id, SQLERRM;
+        END;
+    END IF;
+    RETURN NULL;
+END
+$$;
+
+CREATE CONSTRAINT TRIGGER events_refresh_channel_ttl
+AFTER INSERT ON events
+DEFERRABLE INITIALLY DEFERRED
+FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert();