Skip to content

fix(sync): emit sortAt ops by TARGET name in seconds (source-name ops silently dropped) #334

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

Merged
JustMaier merged 1 commit into from
Jul 18, 2026
Merged

fix(sync): emit sortAt ops by TARGET name in seconds (source-name ops silently dropped) #334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
32 changes: 19 additions & 13 deletions deploy/configs/sync-config-civitai.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,13 +237,19 @@ triggers:
slot_field: id
sets_alive: true
on_delete: true
# [PR-B4] PINNED UNIT STATEMENT (emission + query chains — see also civitai-index.yaml):
# EMISSION: sortAtUnix is emitted in MILLISECONDS (`* 1000`). The engine
# data_schema mapping `sortAtUnix -> sortAt` (civitai-index.yaml,
# ms_to_seconds:true) divides by 1000, so the sortAt SORT LAYERS store
# SECONDS. The dump path instead emits a `sortAt` column already in SECONDS
# (data_schema fallback: sortAt — ms_to_seconds is NOT applied to a
# fallback source). Both wire encodings converge on SECONDS in the layers.
# [PR-B4-v2] PINNED UNIT STATEMENT (emission + query chains — see also civitai-index.yaml):
# EMISSION: ops emit field `sortAt` (the TARGET name) in SECONDS. This was
# CORRECTED 2026-07-18: the original design emitted `sortAtUnix` in ms,
# assuming the data_schema `sortAtUnix -> sortAt` mapping applied to op
# field names — IT DOES NOT. data_schema renames apply to DOCUMENT keys
# (dump/insert path) only; op Set/Remove fields are resolved raw against
# filter/sort TARGET names (ops_processor.rs process_set_op), and unknown
# fields are SILENTLY IGNORED. Every other field always emitted target
# names — sortAtUnix was the one deviation, and its ops were dropped in
# prod for ~5h until caught by a single-op experiment. Target-name seconds
# is now the rule. The dormant sortAtUnix->sortAt doc mapping remains for
# the dump path only. The dump path emits a `sortAt` CSV column in SECONDS
# (doc path — mapping applies). Both paths converge on SECONDS everywhere.
# QUERY: the Civitai builder sends `sortAtUnix Gte <seconds>`
# (image.service.ts:3940). That clause is resolved by the time-bucket
# manager (civitai-index.yaml time_buckets.filter_field: sortAtUnix,
Expand All @@ -256,9 +262,9 @@ triggers:
# divided back on ingest. The builder's seconds-Gte is CORRECT both before
# this change (sortAt computed) and after (sortAt ingested).
#
# sortAtUnix is emitted via the SHARED bitdex_image_sortat_ops() function so
# the W1-3 re-emitter re-asserts the identical value (shape parity). MS units
# (`* 1000`) with a COALESCE(NEW."sortAt", GREATEST(...)) shape — trusting
# sortAt is emitted via the SHARED bitdex_image_sortat_ops() function so
# the W1-3 re-emitter re-asserts the identical value (shape parity). SECONDS
# with a COALESCE(NEW."sortAt", GREATEST(...)) shape — trusting
# NEW."sortAt" here is UNCHANGED from before the no-backfill decision and
# stays COALESCE-shaped, because it is SAFE for a different reason than the
# dump: the model-share `image_sort_at_before` BEFORE INSERT OR UPDATE
Expand All @@ -272,8 +278,8 @@ triggers:
# comment above: ops emission may trust the column ONLY because
# image_sort_at_before recomputes it on every write; the dump must NEVER
# trust it (92M pre-trigger rows are stale). Expression is byte-identical to
# the W1-1 codegen pin (trigger_gen.rs SORTATUNIX_EXPR).
shared_ops_fields: [sortAtUnix]
# the W1-1 codegen pin (trigger_gen.rs SORTAT_EXPR).
shared_ops_fields: [sortAt]
track_fields:
- nsfwLevel
- { column: type, expression: "{type}::text" }
Expand All @@ -284,7 +290,7 @@ triggers:
- hash
- width
- height
- 'COALESCE(extract(epoch from {sortAt})::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = {postId}))::bigint, extract(epoch from {scannedAt})::bigint, extract(epoch from {createdAt})::bigint)) * 1000 as sortAtUnix'
- 'COALESCE(extract(epoch from {sortAt})::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = {postId}))::bigint, extract(epoch from {scannedAt})::bigint, extract(epoch from {createdAt})::bigint)) as sortAt'
computed_fields:
- target: hasMeta
expression: "({flags} >> 13) & 1 = 1 AND ({flags} >> 2) & 1 = 0"
Expand Down
22 changes: 11 additions & 11 deletions docs/design/trigger-sql-review.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,17 +48,17 @@ CREATE TRIGGER trg_cleanup_bitdex_ops
-- Part 2: Per-table triggers (generated from sync config YAML)
-- -----------------------------------------------------------------------

-- [1/8] Table: Image → Trigger: bitdex_image_ee936694
-- [1/8] Table: Image → Trigger: bitdex_image_dda29cf8
-- Sets alive: yes
-- On delete: emit delete op

CREATE OR REPLACE FUNCTION bitdex_image_sortat_ops(_i "Image") RETURNS jsonb AS $$
SELECT jsonb_build_array(
jsonb_build_object('op', 'set', 'field', 'sortAtUnix', 'value', to_jsonb(COALESCE(extract(epoch from _i."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = _i."postId"))::bigint, extract(epoch from _i."scannedAt")::bigint, extract(epoch from _i."createdAt")::bigint)) * 1000))
jsonb_build_object('op', 'set', 'field', 'sortAt', 'value', to_jsonb(COALESCE(extract(epoch from _i."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = _i."postId"))::bigint, extract(epoch from _i."scannedAt")::bigint, extract(epoch from _i."createdAt")::bigint))))
);
$$ LANGUAGE sql STABLE;

CREATE OR REPLACE FUNCTION bitdex_image_ops_ee936694() RETURNS trigger AS $$
CREATE OR REPLACE FUNCTION bitdex_image_ops_dda29cf8() RETURNS trigger AS $$
DECLARE
_ops jsonb;
BEGIN
Expand Down Expand Up @@ -146,10 +146,10 @@ BEGIN
jsonb_build_object('op', 'set', 'field', 'height', 'value', to_jsonb(NEW."height"))
);
END IF;
IF (COALESCE(extract(epoch from OLD."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = OLD."postId"))::bigint, extract(epoch from OLD."scannedAt")::bigint, extract(epoch from OLD."createdAt")::bigint)) * 1000) IS DISTINCT FROM (COALESCE(extract(epoch from NEW."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = NEW."postId"))::bigint, extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint)) * 1000) THEN
IF (COALESCE(extract(epoch from OLD."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = OLD."postId"))::bigint, extract(epoch from OLD."scannedAt")::bigint, extract(epoch from OLD."createdAt")::bigint))) IS DISTINCT FROM (COALESCE(extract(epoch from NEW."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = NEW."postId"))::bigint, extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint))) THEN
_ops := _ops || jsonb_build_array(
jsonb_build_object('op', 'remove', 'field', 'sortAtUnix', 'value', to_jsonb(COALESCE(extract(epoch from OLD."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = OLD."postId"))::bigint, extract(epoch from OLD."scannedAt")::bigint, extract(epoch from OLD."createdAt")::bigint)) * 1000)),
jsonb_build_object('op', 'set', 'field', 'sortAtUnix', 'value', to_jsonb(COALESCE(extract(epoch from NEW."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = NEW."postId"))::bigint, extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint)) * 1000))
jsonb_build_object('op', 'remove', 'field', 'sortAt', 'value', to_jsonb(COALESCE(extract(epoch from OLD."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = OLD."postId"))::bigint, extract(epoch from OLD."scannedAt")::bigint, extract(epoch from OLD."createdAt")::bigint)))),
jsonb_build_object('op', 'set', 'field', 'sortAt', 'value', to_jsonb(COALESCE(extract(epoch from NEW."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = NEW."postId"))::bigint, extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint))))
);
END IF;
IF ((OLD."flags" >> 13) & 1 = 1 AND (OLD."flags" >> 2) & 1 = 0) IS DISTINCT FROM ((NEW."flags" >> 13) & 1 = 1 AND (NEW."flags" >> 2) & 1 = 0) THEN
Expand Down Expand Up @@ -214,10 +214,10 @@ BEGIN
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS bitdex_image_ee936694 ON "Image";
CREATE TRIGGER bitdex_image_ee936694 AFTER INSERT OR UPDATE OR DELETE ON "Image"
FOR EACH ROW EXECUTE FUNCTION bitdex_image_ops_ee936694();
ALTER TABLE "Image" ENABLE ALWAYS TRIGGER bitdex_image_ee936694;
DROP TRIGGER IF EXISTS bitdex_image_dda29cf8 ON "Image";
CREATE TRIGGER bitdex_image_dda29cf8 AFTER INSERT OR UPDATE OR DELETE ON "Image"
FOR EACH ROW EXECUTE FUNCTION bitdex_image_ops_dda29cf8();
ALTER TABLE "Image" ENABLE ALWAYS TRIGGER bitdex_image_dda29cf8;


-- [2/8] Table: TagsOnImageNew → Trigger: bitdex_tagsonimagenew_bcbef3c3
Expand Down Expand Up @@ -494,7 +494,7 @@ ALTER TABLE "Model" ENABLE ALWAYS TRIGGER bitdex_model_a13d0fe3;
-- -----------------------------------------------------------------------
-- Tables created: BitdexOps, bitdex_cursors
-- Triggers: 8
-- bitdex_image_ee936694 on "Image"
-- bitdex_image_dda29cf8 on "Image"
-- bitdex_tagsonimagenew_bcbef3c3 on "TagsOnImageNew"
-- bitdex_imagetool_f87e1fc4 on "ImageTool"
-- bitdex_imagetechnique_ee2b2860 on "ImageTechnique"
Expand Down
63 changes: 36 additions & 27 deletions src/pg_sync/sync_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -840,7 +840,7 @@ triggers: []

/// [W1-1 review minor 3 / PR-m2] Disjointness on the REAL deploy config.
///
/// The load-bearing invariant is that the new `sortAtUnix` writer (the Image
/// The load-bearing invariant is that the new `sortAt` writer (the Image
/// trigger's shared function) and the Post per-image fan-out never emit the
/// SAME field for one image — an overlap would let op_dedup LIFO-resolve the
/// two writers nondeterministically.
Expand All @@ -851,9 +851,9 @@ triggers: []
/// safe: the two writers fire for temporally-separated events and resolve to
/// the SAME value (both read Post.publishedAt), so LIFO order is immaterial.
/// This test pins that the overlap is EXACTLY that documented set and that
/// `sortAtUnix` (and `model3dId`) are strictly disjoint.
/// `sortAt` (and `model3dId`) are strictly disjoint.
#[test]
fn test_post_fanout_disjoint_from_sortatunix() {
fn test_post_fanout_disjoint_from_sortat() {
use crate::pg_sync::trigger_gen::generate_trigger_sql;

let config = load_deploy_sync_config();
Expand All @@ -873,16 +873,20 @@ triggers: []
assert!(!post_fields.is_empty() && !image_fields.is_empty());

// Post fan-out emits exactly the retained payload (PR-M1) — and crucially
// NOT sortAtUnix.
// NOT sortAt.
let expected_post: std::collections::HashSet<String> =
["publishedAt", "availability", "postedToId"].iter().map(|s| s.to_string()).collect();
assert_eq!(post_fields, expected_post,
"Post fan-out must emit exactly {{publishedAt, availability, postedToId}}, got {post_fields:?}");

// sortAtUnix is emitted by the Image trigger and is DISJOINT from Post.
assert!(image_fields.contains("sortAtUnix"), "Image trigger must emit sortAtUnix");
assert!(!post_fields.contains("sortAtUnix"),
"Post fan-out must NOT emit sortAtUnix — it is the Image trigger's field");
// sortAt is emitted by the Image trigger (TARGET name, seconds — the
// 2026-07-18 correction) and is DISJOINT from Post.
assert!(image_fields.contains("sortAt"), "Image trigger must emit sortAt");
assert!(!image_fields.contains("sortAtUnix"),
"Image trigger must NOT emit the source name sortAtUnix — op fields \
resolve against TARGET names only; source-name ops are silently dropped");
assert!(!post_fields.contains("sortAt"),
"Post fan-out must NOT emit sortAt — it is the Image trigger's field");
// model3dId likewise Image-only (INSERT-read, not carried on the fan-out).
assert!(image_fields.contains("model3dId"), "Image trigger must emit model3dId");
assert!(!post_fields.contains("model3dId"), "Post fan-out must NOT emit model3dId");
Expand All @@ -893,28 +897,32 @@ triggers: []
assert_eq!(overlap, expected_post,
"the only Post∩Image overlap must be the intentional Mode-B set \
{{publishedAt, availability, postedToId}}; any other overlap (esp. \
sortAtUnix) would be op_dedup-nondeterministic. Got {overlap:?}");
sortAt) would be op_dedup-nondeterministic. Got {overlap:?}");
}

/// [PR-B4 emission side / AR-4] The Image trigger emits sortAtUnix in
/// MILLISECONDS via the shared function, on the REAL deploy config. Seconds
/// here would shrink every value 1000× once the engine's ms_to_seconds
/// mapping divides by 1000.
/// [PR-B4-v2 emission side] The Image trigger emits the TARGET field name
/// `sortAt` in SECONDS via the shared function, on the REAL deploy config.
/// CORRECTED 2026-07-18: the original sortAtUnix/ms emission was silently
/// dropped by the engine — op field names resolve against filter/sort TARGET
/// names only (data_schema renames are doc-path only), and unknown op fields
/// are ignored without error.
#[test]
fn test_image_sortatunix_emitted_ms_on_deploy_config() {
fn test_image_sortat_emitted_target_name_seconds_on_deploy_config() {
use crate::pg_sync::trigger_gen::generate_trigger_sql;

let config = load_deploy_sync_config();
let image = config.triggers.iter().find(|t| t.table == "Image")
.expect("Image trigger must exist");
assert_eq!(image.shared_ops_fields.as_deref(), Some(&["sortAtUnix".to_string()][..]),
"Image must factor sortAtUnix into the shared re-emitter function");
assert_eq!(image.shared_ops_fields.as_deref(), Some(&["sortAt".to_string()][..]),
"Image must factor sortAt into the shared re-emitter function");

let sql = generate_trigger_sql(image);
assert!(sql.contains("* 1000"), "sortAtUnix must be milliseconds:\n{sql}");
assert!(!sql.contains("* 1000"), "sortAt must be SECONDS, not ms:\n{sql}");
assert!(sql.contains("_ops := _ops || bitdex_image_sortat_ops(NEW);"),
"INSERT must call the shared sortat ops function (re-emitter parity):\n{sql}");
assert!(emitted_trigger_fields(&sql).contains("sortAtUnix"));
assert!(emitted_trigger_fields(&sql).contains("sortAt"));
assert!(!emitted_trigger_fields(&sql).contains("sortAtUnix"),
"source-name emission would be silently dropped by the engine");
}

/// [PR-B4 mapping side + PR-M5] Index config: sortAt is INGESTED not computed,
Expand Down Expand Up @@ -1029,22 +1037,23 @@ triggers: []
assert!(images.fields.iter().any(|f| f.target() == "sortAt"),
"images phase fields must map sortAt");

// PINNED PAIRING: the ops trigger's sortAtUnix expression is UNCHANGED
// and stays COALESCE-shaped — trusting the column there is a *different*
// PINNED PAIRING: the ops trigger's sortAt expression stays
// COALESCE-shaped — trusting the column there is a *different*
// safety argument (image_sort_at_before recomputes on every write), not
// a relaxation of the dump's no-trust rule.
// a relaxation of the dump's no-trust rule. (2026-07-18: emission field
// renamed sortAtUnix -> sortAt / ms -> seconds; the pairing is unchanged.)
let image_trigger = config.triggers.iter().find(|t| t.table == "Image")
.expect("Image trigger must exist");
let sortatunix_track = image_trigger.track_fields.as_deref().unwrap_or(&[])
let sortat_track = image_trigger.track_fields.as_deref().unwrap_or(&[])
.iter()
.map(|tf| tf.to_track_string())
.find(|s| s.contains("sortAtUnix"))
.expect("Image trigger must have a sortAtUnix track field");
.find(|s| s.contains("as sortAt"))
.expect("Image trigger must have a sortAt track field");
assert!(
sortatunix_track.to_uppercase().contains("COALESCE"),
"ops emission sortAtUnix expression must remain COALESCE-shaped \
sortat_track.to_uppercase().contains("COALESCE"),
"ops emission sortAt expression must remain COALESCE-shaped \
(safe via image_sort_at_before recompute-on-write, unlike the \
dump):\n{sortatunix_track}"
dump):\n{sortat_track}"
);
}
}
Loading
Loading