From f8778d14a9b20c2c5669475e7bcaab3d114a6781 Mon Sep 17 00:00:00 2001 From: Hannah Tsukamoto Date: Fri, 31 Jul 2026 11:54:17 -0400 Subject: [PATCH 1/2] feat(cwg): wire DataPipe egress with per-round incremental saves (#3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #3. Adds the save target this file has never had. Until now the only egress was a download button the participant had to choose to click, so every dropout, lobby timeout, and spectator contributed nothing — precisely the sessions needed to characterise attrition. Builds on the CONFIG block and identifiers from the parent branch rather than duplicating them; the save-policy values join that block with their DECISIONS.md rows (A3, A4, B4). The retry classifier is the part worth reviewing. It was written against DataPipe's server source (jspsych/datapipe, functions/src/api-data.ts) rather than the plugin's example snippet, because the semantics decide whether retrying is correct or actively harmful: 201 uploaded. 202 means the OSF upload FAILED but DataPipe persisted the data and queued its own server-side retry — a success for us, and retrying it would duplicate rows. 400 OSF_FILE_EXISTS is the one path that is not queued and where data is genuinely dropped, hence a nonce in every filename. Other 4xx (session limit, validation, unknown experiment) are configuration errors, not transient faults; retrying burns the redirect budget and still loses the data. Only 429/5xx/network are retried, with backoff honouring Retry-After. Classification keys on status class rather than error strings, so upstream adding a new 4xx — the in-flight provider-migration branch adds PROVIDER_NOT_CONNECTED — is handled correctly with no change here. Chunks are disjoint and carry row_range, so reassembly is a concatenation and a duplicate chunk is detectable rather than silently merged. Failed chunks return their rows to the queue for a later flush: duplicates are recoverable in analysis, missing rows are not. The nonce is fixed for the lifetime of a save, so retries within one save reuse the filename and are idempotent. If attempt 1 lands at OSF and only its response is lost, attempt 2 gets OSF_FILE_EXISTS; that now counts as saved, because treating it as failure would requeue the rows and re-send them under a fresh nonce, duplicating them. A collision on the first attempt is a genuine clash with another session and stays terminal. flush() races the save against a redirect budget so no exit can strand a participant waiting on OSF, and pagehide uses keepalive to catch tab-close and connection loss. Wired to the exits that exist: per-round, completion, and spectator. The abort (#5) and no-match (#6) exits are one flush call each once those screens exist. Adds tests/pipeline.test.mjs: 34 checks, no dependencies, no browser. Covers chunking, every response class, requeue-on-failure, concurrent saves, filename uniqueness and fallbacks, the redirect budget, idempotent replay, inert behaviour when unconfigured, and end-to-end reassembly of a partial lossy session. For analysis: chunk_seq may contain gaps, since a failed save consumes a sequence number and its rows reappear later. A gap is not missing data — row_range is the gap-free contiguity check. Co-Authored-By: Claude Opus 5 --- reference-game-cwg.html | 244 ++++++++++++++++++++++++++++++++++++++-- tests/pipeline.test.mjs | 230 +++++++++++++++++++++++++++++++++++++ 2 files changed, 465 insertions(+), 9 deletions(-) create mode 100644 tests/pipeline.test.mjs diff --git a/reference-game-cwg.html b/reference-game-cwg.html index f11c323..e8ca313 100644 --- a/reference-game-cwg.html +++ b/reference-game-cwg.html @@ -97,11 +97,33 @@ ROUND_TIMEOUT_MS: 180000, // --- Data --------------------------------------------------------------------------------- - // TODO(#3): DataPipe experiment ID. Until this is set there is NO data egress except the - // voluntary download button on the final screen, i.e. every participant who does not finish - // contributes nothing. + // TODO(#15): DataPipe experiment ID, from the experiment dashboard. The egress layer below is + // built and tested, but SAVING IS INERT UNTIL THIS IS SET — it warns rather than posting, which + // is safer than a placeholder that uploads to an unintended project. Until it is filled in, + // every participant who does not click the download button contributes nothing. DATAPIPE_EXPERIMENT_ID: "", + // A4 — save per-round chunks rather than once at the end. A single end-of-run save loses + // everything for any dyad that breaks, which is exactly the population #12 needs to + // characterise. Assumes `limitSessions` is OFF (B4): DataPipe's counter increments per SAVE + // CALL, so per-round saving spends ~7 sessions per participant, and a limit sized to + // participants would cap the run a seventh of the way through and drop later saves silently. + SAVE_PER_ROUND: true, + + // A3 — the abort path flushes unconditionally. The one irreversible decision in the register: + // unflushed rows are gone forever, flushed rows can always be ignored in analysis. + FLUSH_ON_ABORT: true, + + // Retry policy. Only transient faults are retried — see classify() for why 202 and 4xx must + // not be, which is not obvious and was checked against DataPipe's source. + SAVE_MAX_ATTEMPTS: 3, + SAVE_BASE_BACKOFF_MS: 500, + + // A save must never hold up a redirect to Prolific (#5, #7). An unpaid participant is a worse + // outcome than a late chunk — and the chunk usually is not lost, since DataPipe queues its own + // retry server-side. + SAVE_REDIRECT_BUDGET_MS: 3000, + // Where the study is hosted, for the Prolific redirect. Kept here so it is obvious that the // production value has to be a real https URL, not a file:// path. PROLIFIC_SUBMIT_URL: "https://app.prolific.com/submissions/complete", @@ -233,8 +255,8 @@ } if (!CONFIG.DATAPIPE_EXPERIMENT_ID) { console.warn( - "[config] No DATAPIPE_EXPERIMENT_ID (#3). There is no data egress except the voluntary " + - "download button — every participant who does not finish will contribute nothing." + "[config] No DATAPIPE_EXPERIMENT_ID (#15). Data egress is wired but INERT — nothing will " + + "be uploaded, so every participant who does not click the download button contributes nothing." ); } if (!PROLIFIC_PID) { @@ -255,6 +277,183 @@ } })(); + // =============================================================================================== + // DATA EGRESS (#3) — DataPipe → OSF, incremental per-round saves + // + // Written against DataPipe's server source (jspsych/datapipe, functions/src/api-data.ts) rather + // than the plugin's example snippet, because the response semantics are genuinely counterintuitive + // and determine whether retrying is correct or actively harmful: + // + // 201 Created — uploaded to OSF. + // 202 Accepted — the OSF upload FAILED, but DataPipe persisted the data and queued it for its + // own server-side retry. This is a SUCCESS for us; retrying would duplicate rows. + // 400 OSF_FILE_EXISTS — filename collided. NOT queued for retry; the one path where data is + // genuinely dropped. Hence the nonce in every filename (A4). + // 400 SESSION_LIMIT_REACHED / INVALID_DATA / EXPERIMENT_NOT_FOUND / PROVIDER_NOT_CONNECTED — + // configuration errors, not transient faults. Retrying burns the redirect + // budget and still loses the data; the fix is on the dashboard (B4). + // 5xx / 429 / network — genuinely transient. The only ones worth retrying. + // + // Classification keys on the STATUS CLASS, not on error strings, so an upstream change that adds + // a new 4xx (the in-flight provider-migration branch adds PROVIDER_NOT_CONNECTED) is handled + // correctly without a change here. + // =============================================================================================== + const DATAPIPE_ENDPOINT = "https://pipe.jspsych.org/api/data/"; + + const Pipeline = (() => { + // Index of the first row not yet included in a chunk. Chunks are disjoint and ordered, so + // reassembly is a concatenation and a duplicate chunk is detectable rather than silently merged. + let savedThrough = 0; + let chunkSeq = 0; + const log = []; + + // DYAD_ID can be null before the adapter has written ?mp_session=; SEED carries a fallback for + // exactly that case, and a filename containing the literal "null" is worth avoiding. + const dyadKey = () => DYAD_ID ?? SEED; + const participantKey = () => PROLIFIC_PID ?? localAdapter?.participantId ?? "local"; + + const configured = () => CONFIG.DATAPIPE_EXPERIMENT_ID.length > 0; + + // Fresh per SAVE, never per attempt and never derived from the round number. Two distinct saves + // must not collide (OSF answers a duplicate with 409, the one unrecoverable failure); the + // retries WITHIN one save must collide, which is what makes them idempotent — see the replay + // case in post(). + const nonce = () => + `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; + + function classify(status) { + if (status === 201 || status === 202) return "ok"; + if (status === 429 || status >= 500) return "retry"; + return "terminal"; + } + + async function post(filename, payload) { + let lastDetail = null; + + for (let attempt = 1; attempt <= CONFIG.SAVE_MAX_ATTEMPTS; attempt++) { + try { + const response = await fetch(DATAPIPE_ENDPOINT, { + method: "POST", + headers: { "Content-Type": "application/json", Accept: "*/*" }, + // keepalive lets an in-flight save survive the page navigating to Prolific. Capped at + // 64KB by spec, which per-round chunks stay well under — another reason A4's chunking + // is load-bearing rather than merely tidy. + keepalive: true, + body: JSON.stringify({ + experimentID: CONFIG.DATAPIPE_EXPERIMENT_ID, + filename, + data: payload, + }), + }); + + const verdict = classify(response.status); + if (verdict === "ok") return { ok: true, status: response.status, attempt }; + + const body = await response.json().catch(() => ({})); + lastDetail = { status: response.status, error: body.error ?? null }; + + // Idempotent replay. Every attempt here posts the SAME filename, so a collision on a + // RETRY means our own earlier attempt did land at OSF and only its response was lost. + // Treating that as failure would requeue the rows and re-send them under a fresh nonce, + // duplicating them — the exact outcome the nonce exists to prevent. A collision on the + // FIRST attempt is a genuine clash with another session and stays terminal. + if (body.error === "OSF_FILE_EXISTS" && attempt > 1) { + console.warn(`[pipeline] ${filename} already landed on an earlier attempt; treating as saved.`); + return { ok: true, status: response.status, attempt, replayed: true }; + } + + if (verdict === "terminal") { + console.error(`[pipeline] ${filename} rejected (terminal):`, lastDetail); + return { ok: false, ...lastDetail, attempt }; + } + + const retryAfter = Number(response.headers.get("Retry-After")); + const backoff = + Number.isFinite(retryAfter) && retryAfter > 0 + ? retryAfter * 1000 + : CONFIG.SAVE_BASE_BACKOFF_MS * 2 ** (attempt - 1); + await new Promise((r) => setTimeout(r, backoff)); + } catch (e) { + // Network fault — indistinguishable from a 5xx here, and treated the same. + lastDetail = { status: null, error: String(e) }; + await new Promise((r) => + setTimeout(r, CONFIG.SAVE_BASE_BACKOFF_MS * 2 ** (attempt - 1)) + ); + } + } + + console.error(`[pipeline] ${filename} exhausted retries:`, lastDetail); + return { ok: false, ...lastDetail, attempt: CONFIG.SAVE_MAX_ATTEMPTS }; + } + + // `label` names the exit path, so a chunk's provenance is legible from its filename alone: + // round-3, abort-partner-dropped, no-match, complete. + async function save(label) { + if (!configured()) { + console.warn(`[pipeline] No DATAPIPE_EXPERIMENT_ID — "${label}" NOT saved (#3).`); + return { ok: false, skipped: true }; + } + + const all = jsPsych.data.get().values(); + const rows = all.slice(savedThrough); + if (rows.length === 0) return { ok: true, empty: true }; + + const seq = chunkSeq++; + // Claim the rows BEFORE awaiting, so a save triggered while this one is in flight (a round + // completing during an abort flush) cannot send the same rows twice. + savedThrough = all.length; + + const filename = `cwg_${dyadKey()}_${participantKey()}_${String(seq).padStart( + 2, + "0" + )}-${label}_${nonce()}.json`; + + const result = await post( + filename, + JSON.stringify({ + dyad_id: dyadKey(), + participant_id: participantKey(), + chunk_seq: seq, + chunk_label: label, + row_range: [savedThrough - rows.length, savedThrough - 1], + saved_at: new Date().toISOString(), + trials: rows, + }) + ); + + if (!result.ok && !result.skipped) { + // Put the rows back so a later flush retries them. The end-of-run flush then acts as a + // safety net for every earlier chunk that failed, at the cost of re-sending some rows under + // a new filename — duplicates are recoverable in analysis, missing rows are not. + savedThrough -= rows.length; + } + + log.push({ filename, label, rows: rows.length, ...result }); + return result; + } + + // Flush for exit paths (A3). Resolves when the save finishes OR when the redirect budget + // expires, whichever comes first — a participant must never be held on a spinner waiting for + // OSF before they can submit (#7). + function flush(label) { + return Promise.race([ + save(label), + new Promise((r) => + setTimeout(() => r({ ok: false, timedOut: true }), CONFIG.SAVE_REDIRECT_BUDGET_MS) + ), + ]); + } + + return { save, flush, log: () => log, configured }; + })(); + + // Last-ditch flush for exits we cannot intercept: tab close, browser back, connection loss. + // `pagehide` fires where `beforeunload` does not (bfcache, mobile Safari), and the save uses + // keepalive so it survives the page going away. Unconditional, per A3. + window.addEventListener("pagehide", () => { + if (CONFIG.FLUSH_ON_ABORT) Pipeline.save("unload"); + }); + // The 12 tangrams are real PNGs, and `rt` is a reported DV — without this the first exposure to // each image loads mid-trial. `stimuli` is an OBJECT parameter, so jsPsych's automatic media // preloading cannot discover the `src` values; they have to be listed explicitly. @@ -309,7 +508,12 @@ stimulus: "

This game is already full. Thanks for your interest!

", choices: "NO_KEYS", trial_duration: 4000, - on_finish: () => jsPsych.multiplayer.disconnect(), + on_finish: () => { + // Flush even here. A spectator has no game data, but their arrival and the fact they were + // turned away is the raw material for the odd-arrival rate #10's waiting room must handle. + if (CONFIG.FLUSH_ON_ABORT) Pipeline.flush("spectator"); + jsPsych.multiplayer.disconnect(); + }, }, ], conditional_function: () => myRole === "spectator", @@ -339,6 +543,12 @@ // the Hawkins file. save_interaction_history: true, round_timeout: CONFIG.ROUND_TIMEOUT_MS, // NOT in the original — see CONFIG and the header + // A4 — flush after every round rather than once at the end. A dyad that breaks in round 4 is + // exactly the case #12 needs data from, and an end-of-run-only save is the shape that loses it. + // Deliberately not awaited: the next round must not wait on OSF. + on_finish: (data) => { + if (CONFIG.SAVE_PER_ROUND) Pipeline.save(`round-${data.round ?? "?"}`); + }, prompt: (role) => role === "director" ? "

You are the Director. Describe your tangrams in badge order (1, 2, 3, …) so your partner can arrange theirs the same way. You can talk back and forth as much as you need.

" @@ -364,6 +574,7 @@ played === 1 ? "" : "s" }.

Thank you — please submit to complete the study.

+

Saving your responses…

${submissionBlockHTML("complete")}

Researcher tools: @@ -372,9 +583,24 @@ choices: "NO_KEYS", on_load: () => { wireSubmissionButton("complete"); - // Retained as a local fallback while there is no save target (#3). Once DataPipe is wired, - // this is a debugging convenience rather than the only egress — it should NOT be the thing a - // participant is relied upon to click. + + // Final flush: the last round's rows, plus anything an earlier chunk failed to send. Fired + // on_load rather than on_finish because this trial takes no keys and never finishes — the + // participant leaves via the submit button. Not awaited before wiring that button: the + // redirect must stay available immediately even if OSF is slow or down (#7). + Pipeline.flush("complete").then((r) => { + const el = document.getElementById("save-status"); + if (!el) return; + el.textContent = r.skipped + ? "" // no egress configured yet (#3) — say nothing rather than something false + : r.ok + ? "Your responses have been saved." + : "Your responses are recorded; upload will be retried automatically."; + }); + + // Retained as a researcher fallback. Now that DataPipe is wired this is a debugging + // convenience rather than the only egress — it is NOT something a participant is relied + // upon to click. document .getElementById("dl") ?.addEventListener("click", () => diff --git a/tests/pipeline.test.mjs b/tests/pipeline.test.mjs new file mode 100644 index 0000000..972fcc3 --- /dev/null +++ b/tests/pipeline.test.mjs @@ -0,0 +1,230 @@ +// Tests for the DataPipe egress module (#3) in reference-game-cwg.html. +// +// node tests/pipeline.test.mjs +// +// No dependencies and no browser. The Pipeline IIFE is extracted from the experiment file and run +// against a fake DataPipe, so the row accounting can be checked directly — that is where a silent +// data-loss bug would hide, and it is not observable from a two-tab smoke test. +// +// The fake distinguishes ATTEMPTS from what actually LANDS at OSF, because a failed POST leaves no +// file behind. Conflating the two hides exactly the bugs this is meant to catch. + +import fs from "fs"; + +const html = fs.readFileSync(new URL("../reference-game-cwg.html", import.meta.url), "utf8"); +const src = html.match(/