Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0bbc656263 | ||
|
|
a98734171c | ||
|
|
a61057cd81 | ||
|
|
55d08e77ed | ||
|
|
1f084b817c | ||
|
|
80050cd2a8 |
+403
-119
@@ -1,21 +1,43 @@
|
||||
import { useEffect, useState } from 'react';
|
||||
import { useEffect, useRef, useState } from 'react';
|
||||
|
||||
// Status returned by GET /api/identity. Private key is never present here —
|
||||
// it is shown exactly once at generation time (see GeneratedKeyPanel).
|
||||
interface IdentityStatus {
|
||||
hasIdentity: boolean;
|
||||
uid?: string;
|
||||
curve?: string;
|
||||
createdAt?: number;
|
||||
}
|
||||
// Admin console for meta-core's account keystore.
|
||||
//
|
||||
// meta-core stores one secp256k1 keypair per file under
|
||||
// {META_CORE_PATH}/identity/accounts/<uid>.json and has NO notion of a current
|
||||
// user — which account is "you" is decided by whichever client signs in
|
||||
// (meta-watch). So this page lists accounts; it never speaks of an active one.
|
||||
//
|
||||
// The private key is returned by exactly two calls: POST /api/identity/generate
|
||||
// (once, at creation) and POST /api/identity/reveal (on demand, for an account
|
||||
// that already exists). Both land in the same exclusive SecretKeyPanel.
|
||||
|
||||
interface GeneratedKey {
|
||||
interface Account {
|
||||
uid: string;
|
||||
curve: string;
|
||||
createdAt: number;
|
||||
privateKeyHex: string;
|
||||
}
|
||||
|
||||
// A secret on screen. Carries no createdAt: /api/identity/reveal answers with
|
||||
// {uid, curve, privateKeyHex} only.
|
||||
interface SecretKeyView {
|
||||
uid: string;
|
||||
privateKeyHex: string;
|
||||
kind: 'generated' | 'revealed';
|
||||
}
|
||||
|
||||
// Which row has an open confirmation, and for what. One discriminated value
|
||||
// rather than a pair of ids, so at most one confirm panel can ever be open.
|
||||
interface Pending {
|
||||
uid: string;
|
||||
kind: 'remove' | 'reveal';
|
||||
}
|
||||
|
||||
// meta-watch's sentinel cell for profile records: display name and avatar are
|
||||
// about the *user*, not a content file, so they live at (uid, "self", ...).
|
||||
// meta-core stores them opaquely and knows nothing of the convention.
|
||||
const PROFILE_CID = 'self';
|
||||
const PROFILE_NAME_KEY = 'profile:name';
|
||||
|
||||
const cardStyle: React.CSSProperties = {
|
||||
background: '#16213e',
|
||||
borderRadius: '8px',
|
||||
@@ -59,6 +81,21 @@ const monoStyle: React.CSSProperties = {
|
||||
borderRadius: '4px',
|
||||
};
|
||||
|
||||
const dangerPanelStyle: React.CSSProperties = {
|
||||
background: '#1f0a0a',
|
||||
border: '1px solid #7f1d1d',
|
||||
padding: '0.75rem 1rem',
|
||||
borderRadius: '4px',
|
||||
marginTop: '0.75rem',
|
||||
};
|
||||
|
||||
const errorTextStyle: React.CSSProperties = {
|
||||
color: '#f87171',
|
||||
background: '#1f0a0a',
|
||||
padding: '0.5rem 0.75rem',
|
||||
borderRadius: '4px',
|
||||
};
|
||||
|
||||
function formatCreatedAt(unixSec?: number): string {
|
||||
if (!unixSec) return '';
|
||||
try {
|
||||
@@ -79,25 +116,76 @@ async function errorMessage(res: Response, fallback: string): Promise<string> {
|
||||
return `${fallback} (HTTP ${res.status})`;
|
||||
}
|
||||
|
||||
function GeneratedKeyPanel({ data, onDismiss }: { data: GeneratedKey; onDismiss: () => void }) {
|
||||
const [copied, setCopied] = useState(false);
|
||||
const copy = async () => {
|
||||
// Copy text, falling back to selecting the element that displays it.
|
||||
//
|
||||
// The fallback is load-bearing, not politeness: the dashboard is served over
|
||||
// plain HTTP, where navigator.clipboard is undefined — without it the copy
|
||||
// buttons do nothing at all on a real deployment.
|
||||
async function copyToClipboard(text: string, fallbackElementId: string): Promise<void> {
|
||||
try {
|
||||
await navigator.clipboard.writeText(data.privateKeyHex);
|
||||
setCopied(true);
|
||||
setTimeout(() => setCopied(false), 2000);
|
||||
await navigator.clipboard.writeText(text);
|
||||
} catch {
|
||||
// Clipboard API blocked (non-HTTPS context, etc.) — fall back to selection.
|
||||
const sel = window.getSelection();
|
||||
const range = document.createRange();
|
||||
const el = document.getElementById('mm-privkey-box');
|
||||
const el = document.getElementById(fallbackElementId);
|
||||
if (sel && el) {
|
||||
const range = document.createRange();
|
||||
range.selectNodeContents(el);
|
||||
sel.removeAllRanges();
|
||||
sel.addRange(range);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function fetchAccounts(): Promise<Account[]> {
|
||||
const res = await fetch('/api/identity/accounts');
|
||||
if (!res.ok) throw new Error(await errorMessage(res, 'Failed to load accounts'));
|
||||
const data = (await res.json()) as { accounts?: Account[] };
|
||||
// Backend order is CreatedAt ascending then uid (identity.List) — stable, so
|
||||
// render it as given rather than re-sorting here.
|
||||
return data.accounts ?? [];
|
||||
}
|
||||
|
||||
// Resolve an account's display name from the User Data Layer, or null.
|
||||
//
|
||||
// Never throws and never reports an error upward: a name is decoration on top
|
||||
// of the uid, and the account list must render identically whether Redis is up
|
||||
// (identity itself is file-backed, the UDL is not). The name is also *unsigned*
|
||||
// — meta-core stores the plaintext beside the signed record and verifies
|
||||
// nothing — so the uid stays on screen as the real identity, and the length cap
|
||||
// bounds what a hostile writer can do to the layout.
|
||||
async function fetchDisplayName(uid: string): Promise<string | null> {
|
||||
const url =
|
||||
`/api/udl/record?uid=${encodeURIComponent(uid)}` +
|
||||
`&cid=${encodeURIComponent(PROFILE_CID)}` +
|
||||
`&key=${encodeURIComponent(PROFILE_NAME_KEY)}`;
|
||||
try {
|
||||
const res = await fetch(url);
|
||||
if (!res.ok) return null; // 503 when storage is down, 500, …
|
||||
const data = (await res.json()) as { exists?: boolean; value?: unknown };
|
||||
// Key off `value`, not `exists`: clearing a name writes a tombstone, which
|
||||
// leaves the cell existing with no plaintext. Private-tier cells look the
|
||||
// same, which is exactly right — there is nothing to show for either.
|
||||
if (typeof data.value !== 'string') return null;
|
||||
const name = data.value.trim();
|
||||
return name ? name.slice(0, 64) : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// The one place a private key is ever displayed. Rendered exclusively (the
|
||||
// account list is hidden while it is open) so the "save it now" contract of a
|
||||
// freshly generated key is not competing with anything, and so the hardcoded
|
||||
// element id below stays unique.
|
||||
function SecretKeyPanel({ data, onDismiss }: { data: SecretKeyView; onDismiss: () => void }) {
|
||||
const [copied, setCopied] = useState(false);
|
||||
|
||||
const copy = async () => {
|
||||
await copyToClipboard(data.privateKeyHex, 'mm-privkey-box');
|
||||
setCopied(true);
|
||||
setTimeout(() => setCopied(false), 2000);
|
||||
};
|
||||
|
||||
return (
|
||||
<div style={{ ...cardStyle, border: '2px solid #dc2626' }}>
|
||||
<div
|
||||
@@ -109,9 +197,19 @@ function GeneratedKeyPanel({ data, onDismiss }: { data: GeneratedKey; onDismiss:
|
||||
marginBottom: '1rem',
|
||||
}}
|
||||
>
|
||||
{data.kind === 'generated' ? (
|
||||
<>
|
||||
<strong>This private key is shown ONCE.</strong> Save it now in a password manager.
|
||||
It will never be displayed again — there is no recovery path. Losing it means losing access
|
||||
to every record signed under this identity.
|
||||
You can ask for it again later with <em>Reveal private key</em>, but only for as long
|
||||
as this account exists on this node — delete it and the key is gone for good.
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<strong>This is the live private key for an existing account.</strong> Anyone holding
|
||||
it can sign as this uid forever, on any machine. There is no revocation and no
|
||||
rotation. Do not paste it into chat, tickets or screenshots.
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
<p style={{ color: '#9ca3af', margin: '0 0 0.5rem' }}>uid (public):</p>
|
||||
<div style={monoStyle}>{data.uid}</div>
|
||||
@@ -124,36 +222,40 @@ function GeneratedKeyPanel({ data, onDismiss }: { data: GeneratedKey; onDismiss:
|
||||
{copied ? 'Copied' : 'Copy private key'}
|
||||
</button>
|
||||
<button style={dangerButtonStyle} onClick={onDismiss}>
|
||||
I have saved it
|
||||
{data.kind === 'generated' ? 'I have saved it' : 'Done — hide it'}
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
function NoIdentityPanel({
|
||||
// Create / import. Always available, not just on an empty keystore: this is an
|
||||
// admin console, so adding an account to a populated node has to be possible.
|
||||
function NewAccountPanel({
|
||||
onGenerated,
|
||||
onImported,
|
||||
}: {
|
||||
onGenerated: (g: GeneratedKey) => void;
|
||||
onGenerated: (s: SecretKeyView) => void;
|
||||
onImported: () => void;
|
||||
}) {
|
||||
const [busy, setBusy] = useState(false);
|
||||
const [importing, setImporting] = useState(false);
|
||||
const [importHex, setImportHex] = useState('');
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
const [notice, setNotice] = useState<string | null>(null);
|
||||
|
||||
const generate = async () => {
|
||||
setBusy(true);
|
||||
setError(null);
|
||||
setNotice(null);
|
||||
try {
|
||||
const res = await fetch('/api/identity/generate', { method: 'POST' });
|
||||
if (!res.ok) {
|
||||
setError(await errorMessage(res, 'Generate failed'));
|
||||
return;
|
||||
}
|
||||
const data = (await res.json()) as GeneratedKey;
|
||||
onGenerated(data);
|
||||
const data = (await res.json()) as { uid: string; privateKeyHex: string };
|
||||
onGenerated({ uid: data.uid, privateKeyHex: data.privateKeyHex, kind: 'generated' });
|
||||
} catch (e) {
|
||||
setError(`Generate failed: ${e instanceof Error ? e.message : String(e)}`);
|
||||
} finally {
|
||||
@@ -165,6 +267,7 @@ function NoIdentityPanel({
|
||||
e.preventDefault();
|
||||
setBusy(true);
|
||||
setError(null);
|
||||
setNotice(null);
|
||||
try {
|
||||
const res = await fetch('/api/identity/import', {
|
||||
method: 'POST',
|
||||
@@ -175,6 +278,15 @@ function NoIdentityPanel({
|
||||
setError(await errorMessage(res, 'Import failed'));
|
||||
return;
|
||||
}
|
||||
// Import is idempotent — presenting the key IS the proof of ownership, so
|
||||
// a key this node already holds is a success, not a conflict. `created`
|
||||
// only decides the wording.
|
||||
const data = (await res.json()) as { created?: boolean };
|
||||
setNotice(
|
||||
data.created
|
||||
? 'Imported — a new account was added to this keystore.'
|
||||
: 'That key was already in this keystore; nothing changed.'
|
||||
);
|
||||
setImportHex('');
|
||||
setImporting(false);
|
||||
onImported();
|
||||
@@ -187,15 +299,16 @@ function NoIdentityPanel({
|
||||
|
||||
return (
|
||||
<div style={cardStyle}>
|
||||
<h3 style={{ marginTop: 0 }}>No identity yet</h3>
|
||||
<p style={{ color: '#9ca3af' }}>
|
||||
The User Data Layer needs a signing key for your ratings, watch progress, and "liked" flags
|
||||
to be attributed to you across devices. Nothing happens automatically — generate one below,
|
||||
or import a key you already have.
|
||||
<h3 style={{ marginTop: 0 }}>Add an account</h3>
|
||||
<p style={{ color: '#9ca3af', marginTop: 0 }}>
|
||||
Generating mints a fresh secp256k1 keypair here. Importing adds a key created elsewhere —
|
||||
it is also how an existing account is moved onto this node, so importing a key the
|
||||
keystore already holds succeeds and changes nothing.
|
||||
</p>
|
||||
{error && (
|
||||
<p style={{ color: '#f87171', background: '#1f0a0a', padding: '0.5rem 0.75rem', borderRadius: '4px' }}>
|
||||
{error}
|
||||
{error && <p style={errorTextStyle}>{error}</p>}
|
||||
{notice && (
|
||||
<p style={{ color: '#9ca3af', background: '#0a0a18', padding: '0.5rem 0.75rem', borderRadius: '4px' }}>
|
||||
{notice}
|
||||
</p>
|
||||
)}
|
||||
<div>
|
||||
@@ -235,71 +348,125 @@ function NoIdentityPanel({
|
||||
);
|
||||
}
|
||||
|
||||
function HasIdentityPanel({ status, onReplace }: { status: IdentityStatus; onReplace: () => void }) {
|
||||
const [confirming, setConfirming] = useState(false);
|
||||
const [busy, setBusy] = useState(false);
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
function AccountCard({
|
||||
account,
|
||||
name,
|
||||
busy,
|
||||
disabled,
|
||||
pending,
|
||||
error,
|
||||
onAsk,
|
||||
onCancel,
|
||||
onReveal,
|
||||
onRemove,
|
||||
}: {
|
||||
account: Account;
|
||||
name?: string;
|
||||
busy: boolean;
|
||||
disabled: boolean;
|
||||
pending: Pending | null;
|
||||
error: string | null;
|
||||
onAsk: (p: Pending) => void;
|
||||
onCancel: () => void;
|
||||
onReveal: (uid: string) => void;
|
||||
onRemove: (uid: string) => void;
|
||||
}) {
|
||||
const [copied, setCopied] = useState(false);
|
||||
const uidBoxId = `mm-uid-${account.uid}`;
|
||||
const label = name ?? account.uid;
|
||||
|
||||
const remove = async () => {
|
||||
setBusy(true);
|
||||
setError(null);
|
||||
try {
|
||||
const res = await fetch('/api/identity?confirm=true', { method: 'DELETE' });
|
||||
if (!res.ok && res.status !== 204) {
|
||||
setError(await errorMessage(res, 'Delete failed'));
|
||||
return;
|
||||
}
|
||||
onReplace();
|
||||
} catch (e) {
|
||||
setError(`Delete failed: ${e instanceof Error ? e.message : String(e)}`);
|
||||
} finally {
|
||||
setBusy(false);
|
||||
setConfirming(false);
|
||||
}
|
||||
const copyUid = async () => {
|
||||
await copyToClipboard(account.uid, uidBoxId);
|
||||
setCopied(true);
|
||||
setTimeout(() => setCopied(false), 2000);
|
||||
};
|
||||
|
||||
return (
|
||||
<div style={cardStyle}>
|
||||
<h3 style={{ marginTop: 0 }}>Active identity</h3>
|
||||
<p style={{ color: '#9ca3af', margin: '0 0 0.5rem' }}>uid:</p>
|
||||
<div style={monoStyle}>{status.uid}</div>
|
||||
<p style={{ color: '#888', marginTop: '1rem' }}>
|
||||
Curve: {status.curve} · Created: {formatCreatedAt(status.createdAt)}
|
||||
<div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'flex-start', gap: '1rem' }}>
|
||||
<div style={{ minWidth: 0, flex: 1 }}>
|
||||
<h3 style={{ margin: '0 0 0.5rem', color: name ? '#fff' : '#888' }}>
|
||||
{name ?? 'Unnamed account'}
|
||||
</h3>
|
||||
<div id={uidBoxId} style={monoStyle}>
|
||||
{account.uid}
|
||||
</div>
|
||||
<p style={{ color: '#888', margin: '0.75rem 0 0', fontSize: '0.85rem' }}>
|
||||
Curve: {account.curve} · Created: {formatCreatedAt(account.createdAt)}
|
||||
</p>
|
||||
<p style={{ color: '#9ca3af', fontSize: '0.85rem' }}>
|
||||
The private key lives at <code>/meta-core/identity/identity.json</code> on this volume and
|
||||
is never returned through the API after generation. Other services on this stack (e.g.
|
||||
meta-watch) sign records by calling <code>POST /api/identity/sign</code>.
|
||||
</p>
|
||||
{error && (
|
||||
<p style={{ color: '#f87171', background: '#1f0a0a', padding: '0.5rem 0.75rem', borderRadius: '4px' }}>
|
||||
{error}
|
||||
</p>
|
||||
)}
|
||||
{!confirming ? (
|
||||
<button style={dangerButtonStyle} onClick={() => setConfirming(true)} disabled={busy}>
|
||||
Replace identity
|
||||
</div>
|
||||
<div style={{ display: 'flex', flexDirection: 'column', gap: '0.5rem', flexShrink: 0 }}>
|
||||
<button style={{ ...buttonStyle, marginRight: 0 }} onClick={copyUid} disabled={disabled}>
|
||||
{copied ? 'Copied' : 'Copy uid'}
|
||||
</button>
|
||||
) : (
|
||||
<div
|
||||
style={{
|
||||
background: '#1f0a0a',
|
||||
border: '1px solid #7f1d1d',
|
||||
padding: '0.75rem 1rem',
|
||||
borderRadius: '4px',
|
||||
marginTop: '0.5rem',
|
||||
}}
|
||||
<button
|
||||
style={{ ...dangerButtonStyle, marginRight: 0 }}
|
||||
onClick={() => onAsk({ uid: account.uid, kind: 'reveal' })}
|
||||
disabled={disabled}
|
||||
>
|
||||
<p style={{ color: '#fecaca', marginTop: 0 }}>
|
||||
Replacing the identity deletes the current keypair from disk. Records signed by the
|
||||
current uid will keep verifying for anyone who has them cached, but this node will no
|
||||
longer be able to write under that uid. <strong>Make sure you have backed up the key
|
||||
first.</strong>
|
||||
</p>
|
||||
<button style={dangerButtonStyle} onClick={remove} disabled={busy}>
|
||||
{busy ? 'Deleting…' : 'Yes, delete identity'}
|
||||
Reveal private key
|
||||
</button>
|
||||
<button style={buttonStyle} onClick={() => setConfirming(false)} disabled={busy}>
|
||||
<button
|
||||
style={{ ...dangerButtonStyle, marginRight: 0 }}
|
||||
onClick={() => onAsk({ uid: account.uid, kind: 'remove' })}
|
||||
disabled={disabled}
|
||||
>
|
||||
Remove
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{error && <p style={{ ...errorTextStyle, marginBottom: 0 }}>{error}</p>}
|
||||
|
||||
{pending?.uid === account.uid && pending.kind === 'reveal' && (
|
||||
<div style={dangerPanelStyle}>
|
||||
<p style={{ color: '#fecaca', marginTop: 0 }}>
|
||||
Show the private key for <strong>{label}</strong>?
|
||||
</p>
|
||||
<p style={{ color: '#fecaca', marginTop: 0 }}>
|
||||
The key <em>is</em> the account. Anyone who copies it can sign records as this uid on
|
||||
any machine, forever — there is no way to revoke or rotate it afterwards. meta-core
|
||||
will hand it over without asking who you are; see the note at the top of this page.
|
||||
</p>
|
||||
<button style={dangerButtonStyle} onClick={() => onReveal(account.uid)} disabled={busy}>
|
||||
{busy ? 'Revealing…' : 'Yes, show the private key'}
|
||||
</button>
|
||||
<button style={buttonStyle} onClick={onCancel} disabled={busy}>
|
||||
Cancel
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{pending?.uid === account.uid && pending.kind === 'remove' && (
|
||||
<div style={dangerPanelStyle}>
|
||||
<p style={{ color: '#fecaca', marginTop: 0 }}>
|
||||
Remove <strong>{label}</strong> from this node?
|
||||
</p>
|
||||
<ul style={{ color: '#fecaca', paddingLeft: '1.2rem', margin: '0 0 0.75rem' }}>
|
||||
<li>
|
||||
<strong>The key is gone for good.</strong> The uid <em>is</em> the public key, and
|
||||
the private half exists only in this node's keystore. If nobody saved it, nothing —
|
||||
not you, not meta-core — can ever sign as this uid again. Reveal and copy it first
|
||||
if there is any doubt.
|
||||
</li>
|
||||
<li>
|
||||
<strong>Nothing is retracted.</strong> Records already signed under this uid keep
|
||||
verifying on every peer that replicated them. Deleting a key is not a recall.
|
||||
</li>
|
||||
<li>
|
||||
<strong>Its data stays, but goes read-only.</strong> The account's User Data Layer
|
||||
cells — likes, ratings, watch progress — survive in Redis, and no newer version of
|
||||
any of them can ever be written.
|
||||
</li>
|
||||
<li>
|
||||
<strong>This signs nobody out elsewhere.</strong> Any device that holds this key
|
||||
still holds it, and can import it back here at any time.
|
||||
</li>
|
||||
</ul>
|
||||
<button style={dangerButtonStyle} onClick={() => onRemove(account.uid)} disabled={busy}>
|
||||
{busy ? 'Removing…' : 'Yes, remove this account'}
|
||||
</button>
|
||||
<button style={buttonStyle} onClick={onCancel} disabled={busy}>
|
||||
Cancel
|
||||
</button>
|
||||
</div>
|
||||
@@ -309,66 +476,183 @@ function HasIdentityPanel({ status, onReplace }: { status: IdentityStatus; onRep
|
||||
}
|
||||
|
||||
export default function Identity() {
|
||||
const [status, setStatus] = useState<IdentityStatus | null>(null);
|
||||
const [generated, setGenerated] = useState<GeneratedKey | null>(null);
|
||||
const [accounts, setAccounts] = useState<Account[] | null>(null);
|
||||
const [names, setNames] = useState<Record<string, string>>({});
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [fetchError, setFetchError] = useState<string | null>(null);
|
||||
const [secret, setSecret] = useState<SecretKeyView | null>(null);
|
||||
const [pending, setPending] = useState<Pending | null>(null);
|
||||
const [busyUid, setBusyUid] = useState<string | null>(null);
|
||||
const [rowError, setRowError] = useState<{ uid: string; message: string } | null>(null);
|
||||
|
||||
// Bumped per refresh so a slow response from a superseded load cannot
|
||||
// overwrite a newer one. The page does not poll (unlike the rest of the
|
||||
// dashboard) — a key list has no business changing under the cursor.
|
||||
const reqGen = useRef(0);
|
||||
|
||||
const refresh = async () => {
|
||||
const gen = ++reqGen.current;
|
||||
setLoading(true);
|
||||
setFetchError(null);
|
||||
let list: Account[];
|
||||
try {
|
||||
const res = await fetch('/api/identity');
|
||||
if (!res.ok) {
|
||||
setFetchError(await errorMessage(res, 'Failed to load identity status'));
|
||||
list = await fetchAccounts();
|
||||
} catch (e) {
|
||||
if (gen !== reqGen.current) return;
|
||||
setFetchError(e instanceof Error ? e.message : String(e));
|
||||
setLoading(false);
|
||||
return;
|
||||
}
|
||||
setStatus((await res.json()) as IdentityStatus);
|
||||
} catch (e) {
|
||||
setFetchError(`Failed to load identity status: ${e instanceof Error ? e.message : String(e)}`);
|
||||
} finally {
|
||||
if (gen !== reqGen.current) return;
|
||||
// Render the list before resolving names, so a down Redis delays nothing.
|
||||
setAccounts(list);
|
||||
setLoading(false);
|
||||
}
|
||||
|
||||
const settled = await Promise.allSettled(list.map((a) => fetchDisplayName(a.uid)));
|
||||
if (gen !== reqGen.current) return;
|
||||
const next: Record<string, string> = {};
|
||||
settled.forEach((r, i) => {
|
||||
if (r.status === 'fulfilled' && r.value) next[list[i].uid] = r.value;
|
||||
});
|
||||
// Replace rather than merge, so a removed account's name goes with it.
|
||||
setNames(next);
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
refresh();
|
||||
}, []);
|
||||
|
||||
const reveal = async (uid: string) => {
|
||||
setBusyUid(uid);
|
||||
setRowError(null);
|
||||
try {
|
||||
// POST with a body rather than ?uid=, so the target stays out of the
|
||||
// proxy's access log. The handler accepts either.
|
||||
const res = await fetch('/api/identity/reveal', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ confirm: true, uid }),
|
||||
});
|
||||
if (!res.ok) {
|
||||
setRowError({ uid, message: await errorMessage(res, 'Reveal failed') });
|
||||
return;
|
||||
}
|
||||
const data = (await res.json()) as { uid: string; privateKeyHex: string };
|
||||
setPending(null);
|
||||
setSecret({ uid: data.uid, privateKeyHex: data.privateKeyHex, kind: 'revealed' });
|
||||
} catch (e) {
|
||||
setRowError({ uid, message: `Reveal failed: ${e instanceof Error ? e.message : String(e)}` });
|
||||
} finally {
|
||||
setBusyUid(null);
|
||||
}
|
||||
};
|
||||
|
||||
const remove = async (uid: string) => {
|
||||
setBusyUid(uid);
|
||||
setRowError(null);
|
||||
try {
|
||||
// The uid is mandatory: with more than one account the backend refuses an
|
||||
// unqualified delete (400 uid_required) rather than guess.
|
||||
const res = await fetch(`/api/identity?uid=${encodeURIComponent(uid)}&confirm=true`, {
|
||||
method: 'DELETE',
|
||||
});
|
||||
if (!res.ok) {
|
||||
setRowError({ uid, message: await errorMessage(res, 'Remove failed') });
|
||||
return;
|
||||
}
|
||||
setPending(null);
|
||||
await refresh();
|
||||
} catch (e) {
|
||||
setRowError({ uid, message: `Remove failed: ${e instanceof Error ? e.message : String(e)}` });
|
||||
} finally {
|
||||
setBusyUid(null);
|
||||
}
|
||||
};
|
||||
|
||||
const dismissSecret = () => {
|
||||
setSecret(null);
|
||||
refresh();
|
||||
};
|
||||
|
||||
const showList = !loading && !fetchError && !secret;
|
||||
|
||||
return (
|
||||
<div style={{ padding: '2rem', maxWidth: '900px', margin: '0 auto' }}>
|
||||
<h1>Identity</h1>
|
||||
<div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'center' }}>
|
||||
<h1 style={{ marginBottom: 0 }}>Identity</h1>
|
||||
<button style={{ ...buttonStyle, marginRight: 0 }} onClick={refresh} disabled={loading}>
|
||||
{loading ? 'Loading…' : 'Refresh'}
|
||||
</button>
|
||||
</div>
|
||||
<p style={{ color: '#9ca3af' }}>
|
||||
Signing key for the User Data Layer (per-user ratings, progress, "liked" flags). One
|
||||
identity per node. The same key on another device makes that device count as "you".
|
||||
Every signing key in this node's keystore. meta-core has no "current user" — it stores
|
||||
accounts, and which one is <em>you</em> is decided by whichever client signs in (e.g.
|
||||
meta-watch). Keys live one JSON file per account under{' '}
|
||||
<code>{'{META_CORE_PATH}'}/identity/accounts/<uid>.json</code>, and are used to sign
|
||||
User Data Layer records (ratings, watch progress, "liked" flags).
|
||||
</p>
|
||||
|
||||
<div style={{ ...dangerPanelStyle, marginTop: 0, marginBottom: '1rem' }}>
|
||||
<p style={{ color: '#fecaca', margin: 0 }}>
|
||||
<strong>Admin page — no authorisation happens here.</strong> <code>/api/identity/*</code>{' '}
|
||||
asks no questions: anything that can reach this API can list, create, reveal or delete{' '}
|
||||
<em>any</em> account's private key. The only access control is the perimeter in front of
|
||||
this port. Client-facing screens (meta-watch) authorise reveal and delete against the
|
||||
caller's own session and expose no delete at all; this page has neither check. Treat
|
||||
reaching it as equivalent to holding every key on this node.
|
||||
</p>
|
||||
</div>
|
||||
|
||||
{loading && <div style={cardStyle}>Loading…</div>}
|
||||
|
||||
{fetchError && (
|
||||
<div style={{ ...cardStyle, border: '1px solid #7f1d1d' }}>
|
||||
<p style={{ color: '#f87171', margin: 0 }}>{fetchError}</p>
|
||||
<p style={{ color: '#f87171', margin: '0 0 1rem' }}>{fetchError}</p>
|
||||
<button style={buttonStyle} onClick={refresh}>
|
||||
Retry
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{!loading && !fetchError && generated && (
|
||||
<GeneratedKeyPanel
|
||||
data={generated}
|
||||
onDismiss={() => {
|
||||
setGenerated(null);
|
||||
refresh();
|
||||
{!loading && !fetchError && secret && <SecretKeyPanel data={secret} onDismiss={dismissSecret} />}
|
||||
|
||||
{showList && (
|
||||
<>
|
||||
<NewAccountPanel onGenerated={(s) => setSecret(s)} onImported={refresh} />
|
||||
|
||||
<h2 style={{ fontSize: '1.1rem', color: '#9ca3af', margin: '1.5rem 0 1rem' }}>
|
||||
Accounts ({accounts?.length ?? 0})
|
||||
</h2>
|
||||
|
||||
{accounts && accounts.length === 0 && (
|
||||
<div style={cardStyle}>
|
||||
<p style={{ color: '#9ca3af', margin: 0 }}>
|
||||
No accounts in this keystore. Accounts are normally created by a client when
|
||||
someone signs up; use <em>Add an account</em> above if you are bootstrapping this
|
||||
node.
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{accounts?.map((a) => (
|
||||
<AccountCard
|
||||
key={a.uid}
|
||||
account={a}
|
||||
name={names[a.uid]}
|
||||
busy={busyUid === a.uid}
|
||||
disabled={busyUid !== null}
|
||||
pending={pending}
|
||||
error={rowError?.uid === a.uid ? rowError.message : null}
|
||||
onAsk={(p) => {
|
||||
setRowError(null);
|
||||
setPending(p);
|
||||
}}
|
||||
onCancel={() => setPending(null)}
|
||||
onReveal={reveal}
|
||||
onRemove={remove}
|
||||
/>
|
||||
)}
|
||||
|
||||
{!loading && !fetchError && !generated && status && !status.hasIdentity && (
|
||||
<NoIdentityPanel onGenerated={(g) => setGenerated(g)} onImported={refresh} />
|
||||
)}
|
||||
|
||||
{!loading && !fetchError && !generated && status && status.hasIdentity && (
|
||||
<HasIdentityPanel status={status} onReplace={refresh} />
|
||||
))}
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
|
||||
@@ -50,6 +50,26 @@ type identityImportResponse struct {
|
||||
UID string `json:"uid"`
|
||||
Curve string `json:"curve"`
|
||||
CreatedAt int64 `json:"createdAt"`
|
||||
// Created distinguishes "this key was new here" (201) from "this key was
|
||||
// already in the keystore" (200). Import is the sign-in primitive for
|
||||
// clients: presenting the secret key is the proof of ownership, so a
|
||||
// re-import must succeed rather than conflict. Clients use the flag only
|
||||
// to word the confirmation ("new profile" vs "welcome back").
|
||||
Created bool `json:"created"`
|
||||
}
|
||||
|
||||
// identityRevealRequest asks for an existing account's private key back. The
|
||||
// key was shown once at generate; without a way to see it again, a client with
|
||||
// a real sign-out has no recovery path for a user who never saved it.
|
||||
type identityRevealRequest struct {
|
||||
Confirm bool `json:"confirm"`
|
||||
UID string `json:"uid"`
|
||||
}
|
||||
|
||||
type identityRevealResponse struct {
|
||||
UID string `json:"uid"`
|
||||
Curve string `json:"curve"`
|
||||
PrivateKeyHex string `json:"privateKeyHex"`
|
||||
}
|
||||
|
||||
type identityDeleteRequest struct {
|
||||
@@ -172,9 +192,17 @@ func (s *Server) handleIdentityImport(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
dir := s.config.IdentityAccountsDir()
|
||||
if identity.ExistsByUID(dir, id.UID) {
|
||||
writeErrorSlug(w, http.StatusConflict, "identity_exists",
|
||||
"an account with this key already exists", false)
|
||||
// Idempotent: importing a key that is already in the keystore is how a
|
||||
// second device signs in to an existing account. Answer 200 with the
|
||||
// stored account (preserving its original createdAt) instead of 409 —
|
||||
// the caller proved ownership by presenting the key.
|
||||
if existing, err := identity.LoadByUID(dir, id.UID); err == nil && existing != nil {
|
||||
writeJSON(w, http.StatusOK, identityImportResponse{
|
||||
UID: existing.UID,
|
||||
Curve: existing.Curve,
|
||||
CreatedAt: existing.CreatedAt,
|
||||
Created: false,
|
||||
})
|
||||
return
|
||||
}
|
||||
if err := identity.SaveAccount(dir, id); err != nil {
|
||||
@@ -185,6 +213,35 @@ func (s *Server) handleIdentityImport(w http.ResponseWriter, r *http.Request) {
|
||||
UID: id.UID,
|
||||
Curve: id.Curve,
|
||||
CreatedAt: id.CreatedAt,
|
||||
Created: true,
|
||||
})
|
||||
}
|
||||
|
||||
// handleIdentityReveal returns an existing account's private key. Deliberately
|
||||
// POST + explicit confirm (never a bare GET) so it cannot be triggered by a
|
||||
// link or an image tag, and so it reads as the deliberate act it is. The
|
||||
// endpoint is auth-gated at the perimeter like the rest of /api/identity/*.
|
||||
func (s *Server) handleIdentityReveal(w http.ResponseWriter, r *http.Request) {
|
||||
var body identityRevealRequest
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
if !body.Confirm && r.URL.Query().Get("confirm") != "true" {
|
||||
writeErrorSlug(w, http.StatusBadRequest, "confirm_required",
|
||||
"revealing a private key requires {\"confirm\": true} or ?confirm=true", false)
|
||||
return
|
||||
}
|
||||
uid := strings.TrimSpace(body.UID)
|
||||
if uid == "" {
|
||||
uid = strings.TrimSpace(r.URL.Query().Get("uid"))
|
||||
}
|
||||
id, status, slug, msg := s.resolveAccount(uid)
|
||||
if status != 0 {
|
||||
writeErrorSlug(w, status, slug, msg, false)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, identityRevealResponse{
|
||||
UID: id.UID,
|
||||
Curve: id.Curve,
|
||||
PrivateKeyHex: id.PrivateKeyHex,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -172,6 +172,7 @@ func (s *Server) setupRoutes() {
|
||||
s.router.HandleFunc("/api/identity/accounts", s.handleIdentityAccounts).Methods("GET")
|
||||
s.router.HandleFunc("/api/identity/generate", s.handleIdentityGenerate).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity/import", s.handleIdentityImport).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity/reveal", s.handleIdentityReveal).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity", s.handleIdentityDelete).Methods("DELETE")
|
||||
s.router.HandleFunc("/api/identity/sign", s.handleIdentitySign).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity/aead-key", s.handleIdentityAEADKey).Methods("GET")
|
||||
@@ -350,6 +351,21 @@ func (s *Server) Start() error {
|
||||
|
||||
// Republish events for all existing metadata
|
||||
go func() {
|
||||
// Build any missing per-record field indexes first, in ONE
|
||||
// keyspace pass. The republish below reads every record, and
|
||||
// an un-indexed record read costs a full-keyspace SCAN — on a
|
||||
// real store that walk warmed at ~1.3 records/s (a ~2.6h tail,
|
||||
// paid again on every restart). Warming first turns the whole
|
||||
// republish into index-served reads. Idempotent, so a
|
||||
// fully-indexed store just pays the single scan.
|
||||
// See storage/field_index.go.
|
||||
started := time.Now()
|
||||
if n, err := s.storage.WarmFieldIndexes(); err != nil {
|
||||
log.Printf("[API] Warning: field index warm failed: %v (reads fall back to SCAN)", err)
|
||||
} else if n > 0 {
|
||||
log.Printf("[API] Warmed %d field indexes in %s", n, time.Since(started).Round(time.Millisecond))
|
||||
}
|
||||
|
||||
count, err := s.RepublishMetadata()
|
||||
if err != nil {
|
||||
log.Printf("[API] Warning: failed to republish metadata on startup: %v", err)
|
||||
|
||||
+31
-3
@@ -27,6 +27,14 @@ type UDLPutRequest struct {
|
||||
Ts int64 `json:"ts"`
|
||||
Record string `json:"record"` // base64 CBOR, opaque + signed
|
||||
Value *json.RawMessage `json:"value,omitempty"`
|
||||
// Tombstone marks this write as a deletion of (uid, cid, key). The record
|
||||
// itself is opaque here — meta-core never decodes the CBOR, and a private-
|
||||
// tier value and a tombstone are equally "no plaintext value" — so the
|
||||
// writer has to say. Without it the cid is never removed from
|
||||
// udl:idx:user:<uid>:key:<key>, which then grows for the life of the
|
||||
// account and is returned in full on every My List / Continue Watching
|
||||
// read. Omitted by older writers, which keeps the previous behaviour.
|
||||
Tombstone bool `json:"tombstone,omitempty"`
|
||||
}
|
||||
|
||||
// UDLPutResponse is the response of PUT /api/udl/record.
|
||||
@@ -40,6 +48,18 @@ type UDLRecordResponse struct {
|
||||
Version int64 `json:"version"`
|
||||
Ts int64 `json:"ts"`
|
||||
Exists bool `json:"exists"`
|
||||
// Value is the plaintext JSON scalar stored beside the opaque record, and
|
||||
// is present only for PUBLIC-tier keys (like, rating, profile:name,
|
||||
// profile:avatar). json.RawMessage rather than string because the field is
|
||||
// generic over keys — a bool for `like`, a number for `rating`, a string
|
||||
// for a profile name (see aggregatePublicValues) — and because it is the
|
||||
// exact type UDLPutRequest.Value uses, so a value written through PUT reads
|
||||
// back byte-identical.
|
||||
//
|
||||
// Absent, never null, when there is no plaintext: private-tier cells,
|
||||
// tombstones and never-written cells must not read as an empty public
|
||||
// value. Readers key off presence, not `exists`.
|
||||
Value *json.RawMessage `json:"value,omitempty"`
|
||||
}
|
||||
|
||||
// UDLAggregate mirrors meta-watch's Aggregate wire shape exactly (snake_case
|
||||
@@ -71,12 +91,20 @@ func (s *Server) handleUDLRecordGet(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, UDLRecordResponse{
|
||||
resp := UDLRecordResponse{
|
||||
Record: cell.Record,
|
||||
Version: cell.Version,
|
||||
Ts: cell.Ts,
|
||||
Exists: exists,
|
||||
})
|
||||
}
|
||||
// json.Valid guards the encoder: a malformed `value` hash field (hand-edited
|
||||
// Redis, a future writer bug) would otherwise fail mid-encode and emit a
|
||||
// truncated body instead of a readable error.
|
||||
if exists && cell.HasValue && json.Valid([]byte(cell.Value)) {
|
||||
raw := json.RawMessage(cell.Value)
|
||||
resp.Value = &raw
|
||||
}
|
||||
writeJSON(w, http.StatusOK, resp)
|
||||
}
|
||||
|
||||
// handleUDLRecordPut handles PUT /api/udl/record. Version-gated: a stale write
|
||||
@@ -104,7 +132,7 @@ func (s *Server) handleUDLRecordPut(w http.ResponseWriter, r *http.Request) {
|
||||
hasValue = true
|
||||
}
|
||||
|
||||
accepted, err := s.storage.UDLUpsertIfNewer(req.Uid, req.Cid, req.Key, req.Version, req.Ts, req.Record, publicValue, hasValue)
|
||||
accepted, err := s.storage.UDLUpsertIfNewer(req.Uid, req.Cid, req.Key, req.Version, req.Ts, req.Record, publicValue, hasValue, req.Tombstone)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/metazla/meta-core/internal/storage"
|
||||
)
|
||||
|
||||
// newUDLTestServer returns a Server wired to an in-memory Redis.
|
||||
//
|
||||
// The struct is built directly rather than through NewServer: NewServer boots
|
||||
// the file watcher, mounts manager, watchers poller and WebDAV handler, none of
|
||||
// which the UDL handlers touch — they use s.storage and nothing else.
|
||||
func newUDLTestServer(t *testing.T) *Server {
|
||||
t.Helper()
|
||||
|
||||
mr := miniredis.RunT(t)
|
||||
c := storage.NewClient("")
|
||||
if err := c.Connect("redis://" + mr.Addr()); err != nil {
|
||||
t.Fatalf("connect to miniredis: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
return &Server{storage: c}
|
||||
}
|
||||
|
||||
// getRecord drives handleUDLRecordGet and decodes the body into raw JSON
|
||||
// fields, so a test can assert on the *encoding* of `value` (is it a quoted
|
||||
// string? is the key there at all?) rather than on what a typed struct would
|
||||
// have coerced it into.
|
||||
func getRecord(t *testing.T, s *Server, uid, cid, key string) (int, map[string]json.RawMessage) {
|
||||
t.Helper()
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/udl/record?uid="+uid+"&cid="+cid+"&key="+key, nil)
|
||||
rr := httptest.NewRecorder()
|
||||
s.handleUDLRecordGet(rr, req)
|
||||
|
||||
var body map[string]json.RawMessage
|
||||
if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("decode response %q: %v", rr.Body.String(), err)
|
||||
}
|
||||
return rr.Code, body
|
||||
}
|
||||
|
||||
// The plaintext value of a PUBLIC-tier cell must come back over the wire, and
|
||||
// come back *as it was written* — a JSON string stays a quoted string, a bool
|
||||
// stays a bool.
|
||||
//
|
||||
// Both halves matter. Without the value at all, the meta-core dashboard can
|
||||
// only ever show 44-character base58 uids: display names live in the User Data
|
||||
// Layer as the public cell (uid, "self", "profile:name"), and the plaintext
|
||||
// sits in Redis where the storage layer has always read it — the handler just
|
||||
// used to drop it. And the type has to stay json.RawMessage: this endpoint is
|
||||
// generic over keys, so a `*string` would mangle `like` (bool) and `rating`
|
||||
// (number) while looking correct for names.
|
||||
func TestUDLRecordGet_ReturnsPublicValue(t *testing.T) {
|
||||
s := newUDLTestServer(t)
|
||||
|
||||
const uid = "zAlice"
|
||||
if ok, err := s.storage.UDLUpsertIfNewer(uid, "self", "profile:name", 1, 100, "cmVj", `"Ada Lovelace"`, true, false); err != nil || !ok {
|
||||
t.Fatalf("seed name: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if ok, err := s.storage.UDLUpsertIfNewer(uid, "bagcsaaa", "like", 1, 100, "cmVj", `true`, true, false); err != nil || !ok {
|
||||
t.Fatalf("seed like: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
code, body := getRecord(t, s, uid, "self", "profile:name")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", code)
|
||||
}
|
||||
if got := string(body["value"]); got != `"Ada Lovelace"` {
|
||||
t.Fatalf("value = %s, want a quoted JSON string", got)
|
||||
}
|
||||
var name string
|
||||
if err := json.Unmarshal(body["value"], &name); err != nil || name != "Ada Lovelace" {
|
||||
t.Fatalf("value should unmarshal to the plain name, got %q err=%v", name, err)
|
||||
}
|
||||
|
||||
// A boolean key survives unquoted — the assertion a *string field fails.
|
||||
if _, body = getRecord(t, s, uid, "bagcsaaa", "like"); string(body["value"]) != "true" {
|
||||
t.Fatalf("like value = %s, want the bare JSON literal true", body["value"])
|
||||
}
|
||||
}
|
||||
|
||||
// A cell with no plaintext must omit `value` entirely rather than send null.
|
||||
//
|
||||
// That is the shape of every private-tier key (seek, watched, card, localpath —
|
||||
// their real value rides encrypted inside the opaque record) and of every
|
||||
// tombstone. A reader keys off presence: "there is no plaintext here" and "the
|
||||
// plaintext is null" are different facts, and a name cleared by its owner
|
||||
// leaves the cell existing with no value, which must not read as a name.
|
||||
func TestUDLRecordGet_OmitsValueForPrivateCell(t *testing.T) {
|
||||
s := newUDLTestServer(t)
|
||||
|
||||
const uid = "zBob"
|
||||
if ok, err := s.storage.UDLUpsertIfNewer(uid, "bagcsaaa", "seek", 1, 100, "ZW5jcnlwdGVk", "", false, false); err != nil || !ok {
|
||||
t.Fatalf("seed private cell: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
code, body := getRecord(t, s, uid, "bagcsaaa", "seek")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", code)
|
||||
}
|
||||
if _, ok := body["value"]; ok {
|
||||
t.Fatalf("private cell must omit `value`, got %s", body["value"])
|
||||
}
|
||||
if string(body["exists"]) != "true" {
|
||||
t.Fatalf("exists = %s, want true", body["exists"])
|
||||
}
|
||||
}
|
||||
|
||||
// A cell that was never written answers 200 with exists:false and no value —
|
||||
// not a 404, and not an empty-string value that would render as a blank name.
|
||||
func TestUDLRecordGet_MissingCellHasNoValue(t *testing.T) {
|
||||
s := newUDLTestServer(t)
|
||||
|
||||
code, body := getRecord(t, s, "zNobody", "self", "profile:name")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", code)
|
||||
}
|
||||
if string(body["exists"]) != "false" {
|
||||
t.Fatalf("exists = %s, want false", body["exists"])
|
||||
}
|
||||
if _, ok := body["value"]; ok {
|
||||
t.Fatalf("missing cell must omit `value`, got %s", body["value"])
|
||||
}
|
||||
}
|
||||
@@ -22,7 +22,7 @@
|
||||
"25": "nzb-posting - content-derived (Message-ID set digest), but not itself fetchable",
|
||||
"20": "BitTorrent infohashes - swarm interop",
|
||||
"10": "midhash256 - the record address",
|
||||
"5": "opaque locators (nzb-release, url) - never outrank a real digest",
|
||||
"5": "opaque locators (nzb-release, url, card) - never outrank a real digest",
|
||||
"0": "weak/unknown digests, unparseable input"
|
||||
},
|
||||
"tieBreak": "lexicographically smaller cid wins",
|
||||
@@ -110,6 +110,12 @@
|
||||
"rank": 5,
|
||||
"note": "codec 0x1006 + IDENTITY multihash 0x00. Same tier/reasoning as nzb-release."
|
||||
},
|
||||
{
|
||||
"name": "card-locator",
|
||||
"cid": "bagdsaaanar2g2zdcor3duojvgq3ts",
|
||||
"rank": 5,
|
||||
"note": "codec 0x1007 + IDENTITY multihash 0x00, digest = varint(len(source)) || source || id — here (\"tmdb\",\"tv:95479\"). Same codec-slot rule as nzb-release/url: match on the CODEC, never the mh. The limit case of the family — the only CID in MetaMesh with NO bytes behind it anywhere, ever: it addresses a *work* (a series, a film), not a file, so the byte path 404s by design and bitswap never offers it (seeding is gated on sha2-256). Tier 5 is load-bearing for the planned cross-source merge: one card record may carry BOTH a TMDB and a MyAnimeList locator, and if a real card body is ever added (see card-tier-search.md §4.3) its sha2-256 must win the canonical election over both locators without any addressing change. Ranking it 0 instead would still lose to a digest, but would put the implementations back out of agreement — which is the exact failure this fixture exists to prevent."
|
||||
},
|
||||
{
|
||||
"name": "sha1",
|
||||
"cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",
|
||||
|
||||
+10
-2
@@ -49,6 +49,14 @@ const (
|
||||
// these — matching on mh would see 0x00 and rank them 0.
|
||||
CodeNzbRelease = 0x1005 // custom — self-describing Newznab release locator
|
||||
CodeURL = 0x1006 // custom — identity-multihash CID wrapping an http(s) URL
|
||||
// CodeCard is the limit case of the family: the only CID with NO bytes
|
||||
// behind it anywhere, ever. It addresses a *work* (a series, a film) via
|
||||
// varint(len(source))‖source‖id — ("tmdb","tv:95479") — so `/api/file/{cid}`
|
||||
// correctly 404s and bitswap never offers it. Ranking it here (rather than
|
||||
// letting it fall to RankUnknown) is what makes the planned "one card, two
|
||||
// CIDs" cross-source merge safe: a TMDB and a MyAnimeList locator can sit on
|
||||
// one record, and a real digest added later still wins the election.
|
||||
CodeCard = 0x1007 // custom — identity-multihash CID wrapping a card (work) id
|
||||
|
||||
// CodeNzbPosting is the INVERSE trap of the two above: its multihash is a
|
||||
// REAL sha2-256 digest (over the normalised, sorted article Message-ID
|
||||
@@ -134,7 +142,7 @@ func Decode(cidStr string) (codec uint64, mhCode uint64, err error) {
|
||||
// swarm interop, not strength.
|
||||
// - midhash256 is the floor among real *digests*: internally meaningful,
|
||||
// externally invisible.
|
||||
// - opaque locators (nzb-release, url) sit BELOW midhash. They are not
|
||||
// - opaque locators (nzb-release, url, card) sit BELOW midhash. They are not
|
||||
// content-derived — two URLs to the same bytes yield two different locator
|
||||
// CIDs — so a locator must never displace a real digest as the identity a
|
||||
// record is advertised under. Tier 5 means a locator wins the election only
|
||||
@@ -157,7 +165,7 @@ func Rank(cidStr string) int {
|
||||
}
|
||||
// Opaque locators. Codec-only: their multihash is identity (0x00), so
|
||||
// there is nothing to match on the mh side.
|
||||
if codec == CodeNzbRelease || codec == CodeURL {
|
||||
if codec == CodeNzbRelease || codec == CodeURL || codec == CodeCard {
|
||||
return RankLocator
|
||||
}
|
||||
// nzb-posting: codec-only, checked BEFORE the mh switch below — its
|
||||
|
||||
@@ -9,6 +9,8 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/metazla/meta-core/internal/storage"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -126,6 +128,14 @@ func (p *MetaPublisher) publishEvent(channel, operation string) {
|
||||
if strings.Contains(key, "__index__") || !strings.HasPrefix(key, "file:") {
|
||||
return
|
||||
}
|
||||
// Skip the per-record field index. It is bookkeeping (see
|
||||
// storage.FieldsField), not a metadata field — publishing it would make
|
||||
// every downstream consumer that parses `file:<id>/<field>` off this
|
||||
// stream (meta-search, meta-dup, meta-fuse, meta-sort) see a phantom
|
||||
// field change on a key they can never read.
|
||||
if strings.HasSuffix(key, "/"+storage.FieldsField) {
|
||||
return
|
||||
}
|
||||
|
||||
// Build event fields
|
||||
fields := map[string]interface{}{
|
||||
|
||||
@@ -83,6 +83,7 @@ func (c *Client) Mint(filePath string, size, mtimeNano int64) (string, error) {
|
||||
pipe.Set(ctx, prefix+"sizeByte", strconv.FormatInt(size, 10), 0)
|
||||
pipe.Set(ctx, prefix+"mtimeNano", strconv.FormatInt(mtimeNano, 10), 0)
|
||||
pipe.SAdd(ctx, c.buildIndexKey(), uuid)
|
||||
c.addFieldsToPipe(ctx, pipe, uuid, []string{"filePath", "sizeByte", "mtimeNano"})
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return "", fmt.Errorf("mint pipeline: %w", err)
|
||||
}
|
||||
@@ -116,7 +117,13 @@ func (c *Client) addAliasLocked(ctx context.Context, uuid, cidStr string) error
|
||||
return fmt.Errorf("uuid and cid are required")
|
||||
}
|
||||
|
||||
keysetKey := c.buildKeyPrefix(uuid) + CIDsKeyPrefix + cidStr
|
||||
keysetField := CIDsKeyPrefix + cidStr
|
||||
keysetKey := c.buildKeyPrefix(uuid) + keysetField
|
||||
|
||||
// Every branch below writes the key-set member, so index the field name
|
||||
// once here — including the two guard branches that skip the alias itself.
|
||||
// Missing it would leave `cids/<cid>` invisible to a record read.
|
||||
defer c.addFieldLocked(ctx, uuid, keysetField)
|
||||
|
||||
// Self-pointing alias guard. If the caller is trying to register
|
||||
// cid:<cid> → <cid> (i.e. the root IS the cid value), that's the
|
||||
@@ -233,26 +240,13 @@ func (c *Client) AddDuplicatePath(uuid, path string) (bool, error) {
|
||||
return added > 0, nil
|
||||
}
|
||||
|
||||
// cidsForRootLocked scans the bare-CID key-set for a root and returns its
|
||||
// members (bare CIDs, with the cids/ prefix stripped). Caller must hold c.mu.
|
||||
// cidsForRootLocked returns a root's bare-CID key-set members (with the cids/
|
||||
// prefix stripped). Caller must hold c.mu.
|
||||
//
|
||||
// Reads the field index and filters by prefix; only an un-indexed root falls
|
||||
// back to a SCAN, and then over the narrower cids/ glob. See field_index.go.
|
||||
func (c *Client) cidsForRootLocked(ctx context.Context, uuid string) ([]string, error) {
|
||||
prefix := c.buildKeyPrefix(uuid) + CIDsKeyPrefix
|
||||
var out []string
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan cids: %w", err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
out = append(out, strings.TrimPrefix(k, prefix))
|
||||
}
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return c.fieldsWithPrefixLocked(ctx, uuid, CIDsKeyPrefix)
|
||||
}
|
||||
|
||||
// DeleteRoot removes everything keyed off uuid: every file:<uuid>/* property
|
||||
|
||||
+97
-71
@@ -109,8 +109,12 @@ func (c *Client) buildIndexKey() string {
|
||||
return c.buildKey("file:__index__")
|
||||
}
|
||||
|
||||
// GetMetadataFlat retrieves all metadata for a file as a flat map
|
||||
// Uses flat keys: SCAN file:{hashId}/* then MGET
|
||||
// GetMetadataFlat retrieves all metadata for a file as a flat map.
|
||||
//
|
||||
// Reads the record's field-name index and MGETs those keys — two round-trips,
|
||||
// independent of keyspace size. Records written before the index existed fall
|
||||
// back to the old SCAN and are indexed on the way out. See field_index.go for
|
||||
// why the index exists and why it is safe to derive.
|
||||
func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
@@ -122,41 +126,7 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
result := make(map[string]string)
|
||||
|
||||
// Scan for all keys with this prefix
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan failed: %w", err)
|
||||
}
|
||||
|
||||
if len(keys) > 0 {
|
||||
values, err := c.client.MGet(ctx, keys...).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("mget failed: %w", err)
|
||||
}
|
||||
for i, key := range keys {
|
||||
field := strings.TrimPrefix(key, prefix)
|
||||
if values[i] != nil {
|
||||
result[field] = values[i].(string)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if len(result) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return result, nil
|
||||
return c.readMetadataLocked(ctx, hashID)
|
||||
}
|
||||
|
||||
// SetMetadataFlat stores metadata for a file using flat keys
|
||||
@@ -179,9 +149,21 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
pipe := c.client.Pipeline()
|
||||
|
||||
fields := make([]string, 0, len(metadata))
|
||||
for field, value := range metadata {
|
||||
pipe.Set(ctx, prefix+field, value, 0)
|
||||
// A payload carrying the reserved name (a hand-made snapshot, a buggy
|
||||
// client) would SET a string over the index SET and break every
|
||||
// subsequent read of this record. Drop it — it is derived state, so
|
||||
// there is never a reason to accept it from outside.
|
||||
if isReservedFieldName(field) {
|
||||
continue
|
||||
}
|
||||
pipe.Set(ctx, prefix+field, value, 0)
|
||||
fields = append(fields, field)
|
||||
}
|
||||
// Same pipeline as the SETs, so indexing the field names costs no extra
|
||||
// round-trip. See field_index.go.
|
||||
c.addFieldsToPipe(ctx, pipe, hashID, fields)
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return fmt.Errorf("pipeline exec failed: %w", err)
|
||||
@@ -256,6 +238,12 @@ func (c *Client) SetProperty(hashID, property, value string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// See SetMetadataFlat: the reserved name addresses the derived index SET,
|
||||
// not a metadata field.
|
||||
if isReservedFieldName(property) {
|
||||
return fmt.Errorf("field name %q is reserved", property)
|
||||
}
|
||||
|
||||
key := c.buildKeyPrefix(hashID) + property
|
||||
|
||||
// Use SET to set the key
|
||||
@@ -263,6 +251,8 @@ func (c *Client) SetProperty(hashID, property, value string) error {
|
||||
return fmt.Errorf("set failed: %w", err)
|
||||
}
|
||||
|
||||
c.addFieldLocked(ctx, hashID, property)
|
||||
|
||||
// Add to index set
|
||||
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||
return fmt.Errorf("sadd index failed: %w", err)
|
||||
@@ -387,32 +377,14 @@ func normalizeFilesRelativePath(p string) string {
|
||||
// getMetadataFlatInternal mirrors GetMetadataFlat without re-acquiring the lock.
|
||||
// Callers must already hold c.mu.
|
||||
func (c *Client) getMetadataFlatInternal(ctx context.Context, hashID string) (map[string]string, error) {
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
result := make(map[string]string)
|
||||
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
result, err := c.readMetadataLocked(ctx, hashID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan failed: %w", err)
|
||||
}
|
||||
if len(keys) > 0 {
|
||||
values, err := c.client.MGet(ctx, keys...).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("mget failed: %w", err)
|
||||
}
|
||||
for i, key := range keys {
|
||||
if values[i] == nil {
|
||||
continue
|
||||
}
|
||||
field := strings.TrimPrefix(key, prefix)
|
||||
result[field] = values[i].(string)
|
||||
}
|
||||
}
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
return nil, err
|
||||
}
|
||||
// Unlike GetMetadataFlat, this variant hands back an empty map rather than
|
||||
// nil for a record with no fields — GetMetadataDocument writes into it.
|
||||
if result == nil {
|
||||
result = make(map[string]string)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -449,9 +421,17 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
pipe := c.client.Pipeline()
|
||||
|
||||
fields := make([]string, 0, len(metadata))
|
||||
for field, value := range metadata {
|
||||
pipe.Set(ctx, prefix+field, value, 0)
|
||||
// See SetMetadataFlat: never let an inbound payload overwrite the
|
||||
// derived index SET with a string.
|
||||
if isReservedFieldName(field) {
|
||||
continue
|
||||
}
|
||||
pipe.Set(ctx, prefix+field, value, 0)
|
||||
fields = append(fields, field)
|
||||
}
|
||||
c.addFieldsToPipe(ctx, pipe, hashID, fields)
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return 0, fmt.Errorf("pipeline exec failed: %w", err)
|
||||
@@ -484,7 +464,11 @@ func (c *Client) DeleteProperty(hashID, property string) error {
|
||||
defer cancel()
|
||||
|
||||
key := c.buildKeyPrefix(hashID) + property
|
||||
return c.client.Del(ctx, key).Err()
|
||||
if err := c.client.Del(ctx, key).Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
c.removeFieldLocked(ctx, hashID, property)
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddToSet adds a value to a set-type field (stored as pipe-delimited string in flat key)
|
||||
@@ -530,6 +514,9 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
|
||||
return false, err
|
||||
}
|
||||
|
||||
// The SET above may have created the key, so the field name is new.
|
||||
c.addFieldLocked(ctx, hashID, property)
|
||||
|
||||
// Add to index set
|
||||
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||
return true, fmt.Errorf("sadd index failed: %w", err)
|
||||
@@ -582,7 +569,12 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
|
||||
|
||||
// Save back (or delete key if empty)
|
||||
if len(newValues) == 0 {
|
||||
return true, c.client.Del(ctx, key).Err()
|
||||
if err := c.client.Del(ctx, key).Err(); err != nil {
|
||||
return true, err
|
||||
}
|
||||
// The key is gone, so the field name must leave the index too.
|
||||
c.removeFieldLocked(ctx, hashID, property)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
newValue := strings.Join(newValues, "|")
|
||||
@@ -764,6 +756,12 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
|
||||
if k == indexKey {
|
||||
continue
|
||||
}
|
||||
// Per-record field indexes live under the same prefix and are SETs
|
||||
// — MGET would error on them, and they are bookkeeping, not
|
||||
// searchable metadata.
|
||||
if strings.HasSuffix(k, "/"+FieldsField) {
|
||||
continue
|
||||
}
|
||||
keyBatch = append(keyBatch, k)
|
||||
if len(keyBatch) >= 1000 {
|
||||
if err := flush(); err != nil {
|
||||
@@ -1039,6 +1037,12 @@ func (c *Client) FindByValue(contains string, fields []string, limit int) ([]Fin
|
||||
if field == "" {
|
||||
continue
|
||||
}
|
||||
// The per-record field index is a SET, so an MGET over those keys
|
||||
// would fail the whole query with WRONGTYPE. It holds field names,
|
||||
// not values, so there is nothing here to search anyway.
|
||||
if isReservedFieldName(field) {
|
||||
continue
|
||||
}
|
||||
if len(matches) >= limit {
|
||||
truncated = true
|
||||
break
|
||||
@@ -1207,7 +1211,17 @@ func (c *Client) SetRawValue(key, value string) error {
|
||||
if t != "none" && t != "string" {
|
||||
return fmt.Errorf("key has incompatible type %q (only string is editable)", t)
|
||||
}
|
||||
return c.client.Set(ctx, full, value, 0).Err()
|
||||
if err := c.client.Set(ctx, full, value, 0).Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
// The raw editor can mint a record property out of band, so it has to keep
|
||||
// the field index honest like every other write path. Non-record keys
|
||||
// (`cid:*`, `udl:*`, `file:__index__`, the index set itself) split false
|
||||
// and are left alone.
|
||||
if hashID, field, ok := c.splitRecordKey(full); ok {
|
||||
c.addFieldLocked(ctx, hashID, field)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteRawKey deletes the exact key. Returns true if the key existed.
|
||||
@@ -1222,10 +1236,16 @@ func (c *Client) DeleteRawKey(key string) (bool, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
n, err := c.client.Del(ctx, c.buildKey(key)).Result()
|
||||
full := c.buildKey(key)
|
||||
n, err := c.client.Del(ctx, full).Result()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if n > 0 {
|
||||
if hashID, field, ok := c.splitRecordKey(full); ok {
|
||||
c.removeFieldLocked(ctx, hashID, field)
|
||||
}
|
||||
}
|
||||
return n > 0, nil
|
||||
}
|
||||
|
||||
@@ -1257,14 +1277,20 @@ func (c *Client) IndexRemoveIfEmpty(hashID string) (bool, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
keys, _, err := c.client.Scan(ctx, 0, prefix+"*", 1).Result()
|
||||
nonEmpty, err := c.recordHasFieldsLocked(ctx, hashID)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("scan failed: %w", err)
|
||||
return false, err
|
||||
}
|
||||
if len(keys) > 0 {
|
||||
if nonEmpty {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Nothing left but (possibly) an empty field index — drop it too, so a
|
||||
// de-indexed root leaves no orphan key behind.
|
||||
if err := c.client.Del(ctx, c.buildFieldsKey(hashID)).Err(); err != nil {
|
||||
return false, fmt.Errorf("del field index: %w", err)
|
||||
}
|
||||
|
||||
removed, err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Result()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("srem failed: %w", err)
|
||||
|
||||
@@ -0,0 +1,486 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// The per-record field-name index.
|
||||
//
|
||||
// A record is stored as one flat Redis STRING per leaf
|
||||
// (`file:<uuid>/<field>`), and that shape is deliberate — it is what lets
|
||||
// keyspace notifications fire per *field*, so meta-fuse / meta-editor /
|
||||
// meta-search can react to one field changing instead of diffing a whole hash
|
||||
// (see docs/project-architecture/persistent-storage.md, and
|
||||
// dev/test/suites/flat-keys.bats which asserts records are never hashes).
|
||||
//
|
||||
// The cost of that shape was that reading one record meant asking Redis "which
|
||||
// keys start with `file:<uuid>/`?", and Redis has no prefix index — `SCAN
|
||||
// MATCH` walks the ENTIRE keyspace. On a real store (445k keys, ~36 fields per
|
||||
// record) a single record read cost ~670 SCAN round-trips and ~0.5s of Redis
|
||||
// CPU; SCAN accounted for ~97.5% of all Redis CPU time on the box. That sat on
|
||||
// the byte path for every poster meta-share served, which is how it surfaced.
|
||||
//
|
||||
// So we keep the flat keys and index their *names*:
|
||||
//
|
||||
// file:<uuid>/__fields__ SET{ "filePath", "sizeByte", "cids/bafk…", … }
|
||||
//
|
||||
// A read becomes SMEMBERS + a chunked MGET — two round-trips instead of ~670.
|
||||
// This mirrors `file:__index__` (the set of every root) one level down, and the
|
||||
// bulk reads already worked this way: `GetTuplesForAllFiles` composes its keys
|
||||
// from the index rather than scanning per root.
|
||||
//
|
||||
// **The set is derived, never authoritative.** The flat keys remain the source
|
||||
// of truth. A record whose set is missing (anything written before this landed)
|
||||
// falls back to the old SCAN and populates the set on the way out, so existing
|
||||
// deployments heal themselves on first touch with no migration and no
|
||||
// SchemaVersion bump — which matters, because a bump is a flag day that makes
|
||||
// operators wipe Redis (see schema_version.go).
|
||||
//
|
||||
// The consistency argument that makes a derived index safe here is meta-core's
|
||||
// single-writer rule: leader election publishes an empty `redisUrl` on purpose
|
||||
// (internal/leader/election.go), so nothing outside this process can write a
|
||||
// `file:` key behind the index's back.
|
||||
|
||||
// FieldsField is the reserved field name holding a record's field-name index.
|
||||
//
|
||||
// It is a Redis SET living at `file:<uuid>/__fields__`, i.e. inside the same
|
||||
// prefix as the record's own fields. That keeps every existing sweep correct
|
||||
// for free — `DeleteMetadata`, `DeleteRoot` and `ClearAllMetadata` all
|
||||
// SCAN+DEL `file:<uuid>/*` and therefore drop the index along with the record.
|
||||
// The flip side is that every reader which enumerates that prefix has to skip
|
||||
// it, exactly as they already skip `file:__index__`.
|
||||
const FieldsField = "__fields__"
|
||||
|
||||
// mgetBatch caps how many keys go into one MGET. Redis limits arguments per
|
||||
// command; 1000 is the batch size `GetTuplesForAllFiles` already uses.
|
||||
const mgetBatch = 1000
|
||||
|
||||
// buildFieldsKey constructs the key for a record's field-name index set.
|
||||
func (c *Client) buildFieldsKey(hashID string) string {
|
||||
return c.buildKeyPrefix(hashID) + FieldsField
|
||||
}
|
||||
|
||||
// isReservedFieldName reports whether a field name is bookkeeping rather than
|
||||
// metadata, and so must never be returned to an API caller, exported into a
|
||||
// snapshot, or folded into the search haystack.
|
||||
func isReservedFieldName(field string) bool {
|
||||
return field == FieldsField
|
||||
}
|
||||
|
||||
// readMetadataLocked returns a record's full flat metadata. Caller must hold
|
||||
// c.mu (either mode).
|
||||
//
|
||||
// Fast path: SMEMBERS the field index, then MGET those keys. Fallback: the
|
||||
// legacy full-keyspace SCAN, whose result is used to populate the index so the
|
||||
// next read takes the fast path.
|
||||
//
|
||||
// Returns (nil, nil) for a record with no fields — callers treat that as 404.
|
||||
func (c *Client) readMetadataLocked(ctx context.Context, hashID string) (map[string]string, error) {
|
||||
fields, err := c.client.SMembers(ctx, c.buildFieldsKey(hashID)).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return nil, fmt.Errorf("smembers fields: %w", err)
|
||||
}
|
||||
|
||||
if len(fields) > 0 {
|
||||
return c.mgetFieldsLocked(ctx, hashID, fields)
|
||||
}
|
||||
|
||||
// No index yet: a record written before the index existed, or one that
|
||||
// genuinely has no fields. Fall back to the scan, then backfill.
|
||||
result, scanned, err := c.scanMetadataLocked(ctx, hashID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(scanned) > 0 {
|
||||
// Best-effort: a failed backfill costs the next read a scan, it does
|
||||
// not make the read wrong. Never fail the caller's read for it.
|
||||
if err := c.client.SAdd(ctx, c.buildFieldsKey(hashID), toAny(scanned)...).Err(); err != nil {
|
||||
logFieldIndexBackfillFailure(hashID, err)
|
||||
}
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// mgetFieldsLocked reads the named fields of a record in chunked MGETs.
|
||||
//
|
||||
// A field named in the index whose key is gone reads back as nil and is simply
|
||||
// skipped: a *stale* index entry is harmless. (A *missing* one would silently
|
||||
// drop data, which is why every write path maintains the set.)
|
||||
func (c *Client) mgetFieldsLocked(ctx context.Context, hashID string, fields []string) (map[string]string, error) {
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
|
||||
// Keep the reserved name out of the key list AND out of the result — it is
|
||||
// a SET, so MGET would error on it in real Redis rather than return nil.
|
||||
wanted := make([]string, 0, len(fields))
|
||||
for _, f := range fields {
|
||||
if !isReservedFieldName(f) {
|
||||
wanted = append(wanted, f)
|
||||
}
|
||||
}
|
||||
if len(wanted) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
result := make(map[string]string, len(wanted))
|
||||
for start := 0; start < len(wanted); start += mgetBatch {
|
||||
end := start + mgetBatch
|
||||
if end > len(wanted) {
|
||||
end = len(wanted)
|
||||
}
|
||||
chunk := wanted[start:end]
|
||||
keys := make([]string, len(chunk))
|
||||
for i, f := range chunk {
|
||||
keys[i] = prefix + f
|
||||
}
|
||||
values, err := c.client.MGet(ctx, keys...).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("mget failed: %w", err)
|
||||
}
|
||||
for i, v := range values {
|
||||
if s, ok := v.(string); ok {
|
||||
result[chunk[i]] = s
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(result) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// scanMetadataLocked is the pre-index read: SCAN the record's key prefix, then
|
||||
// MGET. Retained as the fallback for un-indexed records and as the source for
|
||||
// the backfill. Returns the metadata plus the field names it saw (which
|
||||
// includes fields whose value came back nil, so the index stays complete).
|
||||
func (c *Client) scanMetadataLocked(ctx context.Context, hashID string) (map[string]string, []string, error) {
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
result := make(map[string]string)
|
||||
var fields []string
|
||||
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("scan failed: %w", err)
|
||||
}
|
||||
|
||||
if len(keys) > 0 {
|
||||
wanted := make([]string, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
field := strings.TrimPrefix(key, prefix)
|
||||
if isReservedFieldName(field) {
|
||||
continue
|
||||
}
|
||||
fields = append(fields, field)
|
||||
wanted = append(wanted, key)
|
||||
}
|
||||
if len(wanted) > 0 {
|
||||
values, err := c.client.MGet(ctx, wanted...).Result()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("mget failed: %w", err)
|
||||
}
|
||||
for i, key := range wanted {
|
||||
if s, ok := values[i].(string); ok {
|
||||
result[strings.TrimPrefix(key, prefix)] = s
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return result, fields, nil
|
||||
}
|
||||
|
||||
// addFieldsToPipe records field names in the index as part of an existing
|
||||
// pipeline, so a multi-field write stays one round-trip.
|
||||
func (c *Client) addFieldsToPipe(ctx context.Context, pipe redis.Pipeliner, hashID string, fields []string) {
|
||||
names := filterReserved(fields)
|
||||
if len(names) == 0 {
|
||||
return
|
||||
}
|
||||
pipe.SAdd(ctx, c.buildFieldsKey(hashID), toAny(names)...)
|
||||
}
|
||||
|
||||
// addFieldLocked records a single field name in the index. Caller must hold
|
||||
// c.mu. Best-effort by the same argument as the backfill: the flat key is the
|
||||
// source of truth, and a lost index entry is repaired by the next scan
|
||||
// fallback — so it must not fail the write it accompanies.
|
||||
func (c *Client) addFieldLocked(ctx context.Context, hashID, field string) {
|
||||
if isReservedFieldName(field) {
|
||||
return
|
||||
}
|
||||
if err := c.client.SAdd(ctx, c.buildFieldsKey(hashID), field).Err(); err != nil {
|
||||
logFieldIndexWriteFailure(hashID, field, err)
|
||||
}
|
||||
}
|
||||
|
||||
// removeFieldLocked drops a field name from the index. Caller must hold c.mu.
|
||||
//
|
||||
// A stale entry left behind by a failure is harmless (the MGET reads nil and
|
||||
// skips it), so this is best-effort too.
|
||||
func (c *Client) removeFieldLocked(ctx context.Context, hashID, field string) {
|
||||
if isReservedFieldName(field) {
|
||||
return
|
||||
}
|
||||
if err := c.client.SRem(ctx, c.buildFieldsKey(hashID), field).Err(); err != nil {
|
||||
logFieldIndexWriteFailure(hashID, field, err)
|
||||
}
|
||||
}
|
||||
|
||||
// fieldsWithPrefixLocked returns the record's indexed field names under a
|
||||
// field-name prefix (e.g. "cids/"), falling back to a SCAN when the record has
|
||||
// no index yet. Caller must hold c.mu.
|
||||
func (c *Client) fieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix string) ([]string, error) {
|
||||
fields, err := c.client.SMembers(ctx, c.buildFieldsKey(hashID)).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return nil, fmt.Errorf("smembers fields: %w", err)
|
||||
}
|
||||
if len(fields) == 0 {
|
||||
// Un-indexed record: scan the narrower `<prefix><fieldPrefix>*` glob
|
||||
// rather than the whole record, and do not backfill from it — a
|
||||
// partial view must never be mistaken for a complete index.
|
||||
return c.scanFieldsWithPrefixLocked(ctx, hashID, fieldPrefix)
|
||||
}
|
||||
|
||||
var out []string
|
||||
for _, f := range fields {
|
||||
if isReservedFieldName(f) {
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(f, fieldPrefix) {
|
||||
out = append(out, strings.TrimPrefix(f, fieldPrefix))
|
||||
}
|
||||
}
|
||||
// SMEMBERS order is arbitrary and SCAN order was too, so the CID array in
|
||||
// GetMetadataDocument used to shuffle between calls. Sorting costs nothing
|
||||
// at these sizes and makes the response byte-stable, which matters for
|
||||
// anything diffing or caching it.
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *Client) scanFieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix string) ([]string, error) {
|
||||
prefix := c.buildKeyPrefix(hashID) + fieldPrefix
|
||||
var out []string
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan fields: %w", err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
out = append(out, strings.TrimPrefix(k, prefix))
|
||||
}
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// WarmFieldIndexes builds every missing field index in a SINGLE keyspace pass.
|
||||
//
|
||||
// Without it the indexes still appear — each record's first read falls back to
|
||||
// a scan and backfills — but that transition is O(records x keyspace) and it
|
||||
// runs again from scratch after any wipe. Measured on the real store (12,345
|
||||
// records / 445k keys) the lazy path warmed at ~1.3 records/s: a ~2.6 HOUR tail
|
||||
// during which meta-core stays pinned exactly as it was before the fix. Worse,
|
||||
// `RepublishMetadata` walks every record at startup, so that tail is paid on
|
||||
// every restart.
|
||||
//
|
||||
// One bulk `file:*` scan grouped by record collapses it to a few seconds. This
|
||||
// is the same shape `GetSearchIndexTuples` already uses for the bulk read, and
|
||||
// for the same reason.
|
||||
//
|
||||
// Idempotent, and safe to run against a partially-indexed store: SADD of a
|
||||
// member that is already there is a no-op. Returns the number of records it
|
||||
// wrote an index for.
|
||||
func (c *Client) WarmFieldIndexes() (int, error) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
if c.client == nil {
|
||||
return 0, fmt.Errorf("not connected")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
filePrefix := c.buildKey("file:") // {filePrefix}{hashID}/{field}
|
||||
indexKey := c.buildIndexKey()
|
||||
|
||||
byRecord := make(map[string][]string)
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, filePrefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("scan: %w", err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
if k == indexKey {
|
||||
continue
|
||||
}
|
||||
// hashID has no '/'; field may contain them.
|
||||
rest := strings.TrimPrefix(k, filePrefix)
|
||||
slash := strings.IndexByte(rest, '/')
|
||||
if slash < 0 {
|
||||
continue
|
||||
}
|
||||
hashID, field := rest[:slash], rest[slash+1:]
|
||||
if isReservedFieldName(field) {
|
||||
// Already indexed — remember the record so we don't treat it
|
||||
// as missing, but there is nothing to add for this key.
|
||||
if _, seen := byRecord[hashID]; !seen {
|
||||
byRecord[hashID] = nil
|
||||
}
|
||||
continue
|
||||
}
|
||||
byRecord[hashID] = append(byRecord[hashID], field)
|
||||
}
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
written := 0
|
||||
pipe := c.client.Pipeline()
|
||||
pending := 0
|
||||
for hashID, fields := range byRecord {
|
||||
if len(fields) == 0 {
|
||||
continue
|
||||
}
|
||||
pipe.SAdd(ctx, c.buildFieldsKey(hashID), toAny(fields)...)
|
||||
written++
|
||||
pending++
|
||||
// Bound the pipeline so a large store doesn't buffer every command.
|
||||
if pending >= 500 {
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return written, fmt.Errorf("sadd pipeline: %w", err)
|
||||
}
|
||||
pipe = c.client.Pipeline()
|
||||
pending = 0
|
||||
}
|
||||
}
|
||||
if pending > 0 {
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return written, fmt.Errorf("sadd pipeline: %w", err)
|
||||
}
|
||||
}
|
||||
return written, nil
|
||||
}
|
||||
|
||||
// recordHasFieldsLocked reports whether a record still has at least one
|
||||
// property key. Caller must hold c.mu.
|
||||
//
|
||||
// `__fields__` must not count — it lives under the record's own prefix, so a
|
||||
// naive "does any `file:<id>/*` key exist?" probe would answer yes for a record
|
||||
// whose every property has been deleted, and the root could never be
|
||||
// de-indexed.
|
||||
func (c *Client) recordHasFieldsLocked(ctx context.Context, hashID string) (bool, error) {
|
||||
n, err := c.client.SCard(ctx, c.buildFieldsKey(hashID)).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return false, fmt.Errorf("scard fields: %w", err)
|
||||
}
|
||||
if n > 0 {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Either genuinely empty, or an un-indexed record from before the index
|
||||
// existed — the scan is the only way to tell those apart.
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("scan failed: %w", err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
if !isReservedFieldName(strings.TrimPrefix(k, prefix)) {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// splitRecordKey decomposes a full Redis key into (hashID, field) when it is a
|
||||
// record property key. Reports ok=false for anything else — the index set
|
||||
// itself, `file:__index__`, and any key outside the `file:` namespace.
|
||||
//
|
||||
// This is what lets the raw KV escape hatch (SetRawValue / DeleteRawKey) keep
|
||||
// the index honest without its callers knowing the layout.
|
||||
func (c *Client) splitRecordKey(key string) (hashID, field string, ok bool) {
|
||||
rest := key
|
||||
if c.prefix != "" {
|
||||
var found bool
|
||||
rest, found = strings.CutPrefix(rest, c.prefix)
|
||||
if !found {
|
||||
return "", "", false
|
||||
}
|
||||
}
|
||||
rest, found := strings.CutPrefix(rest, "file:")
|
||||
if !found {
|
||||
return "", "", false
|
||||
}
|
||||
hashID, field, found = strings.Cut(rest, "/")
|
||||
if !found || hashID == "" || field == "" {
|
||||
// `file:__index__` has no slash and lands here.
|
||||
return "", "", false
|
||||
}
|
||||
if isReservedFieldName(field) {
|
||||
return "", "", false
|
||||
}
|
||||
return hashID, field, true
|
||||
}
|
||||
|
||||
// The index is derived and self-healing, so a write failure degrades
|
||||
// performance (the next read falls back to a scan and re-populates) rather
|
||||
// than correctness. Log it rather than failing the caller's write — but log it
|
||||
// loudly, because a *persistent* failure means every read is scanning again.
|
||||
func logFieldIndexWriteFailure(hashID, field string, err error) {
|
||||
log.Printf("[Storage] field index update failed for %s/%s: %v (reads fall back to SCAN)", hashID, field, err)
|
||||
}
|
||||
|
||||
func logFieldIndexBackfillFailure(hashID string, err error) {
|
||||
log.Printf("[Storage] field index backfill failed for %s: %v (reads keep falling back to SCAN)", hashID, err)
|
||||
}
|
||||
|
||||
func filterReserved(fields []string) []string {
|
||||
out := make([]string, 0, len(fields))
|
||||
for _, f := range fields {
|
||||
if !isReservedFieldName(f) {
|
||||
out = append(out, f)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func toAny(ss []string) []interface{} {
|
||||
out := make([]interface{}, len(ss))
|
||||
for i, s := range ss {
|
||||
out[i] = s
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,464 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"sort"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Tests for the per-record field-name index (field_index.go).
|
||||
//
|
||||
// The index is what makes a record read O(fields) instead of O(keyspace). It is
|
||||
// derived state, so the properties worth pinning are: (1) a read served from
|
||||
// the index returns exactly what a scan would have, (2) a record with no index
|
||||
// still reads correctly and gets one, and (3) every write path keeps it in
|
||||
// sync — because a *missing* entry silently drops a field from every
|
||||
// subsequent read, which is the one failure mode that would be invisible.
|
||||
//
|
||||
// Every case runs with and without a namespace prefix, matching clear_test.go.
|
||||
|
||||
func forEachPrefix(t *testing.T, fn func(t *testing.T, c *Client)) {
|
||||
t.Helper()
|
||||
for _, prefix := range []string{"", "mm:"} {
|
||||
name := "no-prefix"
|
||||
if prefix != "" {
|
||||
name = "prefix-" + prefix
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
c, _ := newTestClient(t, prefix)
|
||||
fn(t, c)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func fieldNames(t *testing.T, c *Client, hashID string) []string {
|
||||
t.Helper()
|
||||
got, err := c.client.SMembers(context.Background(), c.buildFieldsKey(hashID)).Result()
|
||||
if err != nil {
|
||||
t.Fatalf("smembers fields: %v", err)
|
||||
}
|
||||
sort.Strings(got)
|
||||
return got
|
||||
}
|
||||
|
||||
func mustGetMetadata(t *testing.T, c *Client, hashID string) map[string]string {
|
||||
t.Helper()
|
||||
got, err := c.GetMetadataFlat(hashID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetMetadataFlat: %v", err)
|
||||
}
|
||||
return got
|
||||
}
|
||||
|
||||
// SetMetadataFlat must index every field it writes, and the read must come back
|
||||
// through the index rather than a scan.
|
||||
func TestFieldIndex_SetMetadataFlatIndexesFields(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
want := map[string]string{
|
||||
"filePath": "/files/watch/a.mkv",
|
||||
"sizeByte": "123",
|
||||
"cids/bafkreiaaa": "true",
|
||||
"titles/jpn": "ナルト",
|
||||
"categories/TV/UHD": "true",
|
||||
"info/files/0/path": "a.mkv",
|
||||
"provenance": `{"source":"gateway"}`,
|
||||
"plot/eng": "a plot with / slashes",
|
||||
"stream/3": "eng",
|
||||
"genres/action": "true",
|
||||
"originaltitle": "Naruto",
|
||||
"deeply/nested/field/name/here": "x",
|
||||
}
|
||||
if err := c.SetMetadataFlat("root1", want); err != nil {
|
||||
t.Fatalf("SetMetadataFlat: %v", err)
|
||||
}
|
||||
|
||||
if got := mustGetMetadata(t, c, "root1"); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("GetMetadataFlat mismatch\n got %v\nwant %v", got, want)
|
||||
}
|
||||
|
||||
// The index must name every field — including the ones containing '/',
|
||||
// which is where a naive "split on the first slash" would go wrong.
|
||||
wantNames := make([]string, 0, len(want))
|
||||
for k := range want {
|
||||
wantNames = append(wantNames, k)
|
||||
}
|
||||
sort.Strings(wantNames)
|
||||
if got := fieldNames(t, c, "root1"); !reflect.DeepEqual(got, wantNames) {
|
||||
t.Errorf("field index mismatch\n got %v\nwant %v", got, wantNames)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A record written before the index existed has flat keys but no set. The read
|
||||
// must still be correct, and must leave an index behind so the next read takes
|
||||
// the fast path. This is the migration path for the ~12k records already on
|
||||
// disk: no flag day, no SchemaVersion bump.
|
||||
func TestFieldIndex_LazyBackfillForUnindexedRecord(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
ctx := context.Background()
|
||||
prefix := c.buildKeyPrefix("legacy")
|
||||
want := map[string]string{
|
||||
"filePath": "/files/watch/old.mkv",
|
||||
"sizeByte": "42",
|
||||
"cids/bafkreiold": "true",
|
||||
}
|
||||
// Write the flat keys directly, bypassing every index-maintaining path
|
||||
// — exactly the shape a pre-upgrade store is in.
|
||||
for f, v := range want {
|
||||
if err := c.client.Set(ctx, prefix+f, v, 0).Err(); err != nil {
|
||||
t.Fatalf("seed %s: %v", f, err)
|
||||
}
|
||||
}
|
||||
if names := fieldNames(t, c, "legacy"); len(names) != 0 {
|
||||
t.Fatalf("precondition: expected no index, got %v", names)
|
||||
}
|
||||
|
||||
if got := mustGetMetadata(t, c, "legacy"); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("scan-fallback read mismatch\n got %v\nwant %v", got, want)
|
||||
}
|
||||
|
||||
wantNames := []string{"cids/bafkreiold", "filePath", "sizeByte"}
|
||||
if got := fieldNames(t, c, "legacy"); !reflect.DeepEqual(got, wantNames) {
|
||||
t.Errorf("backfill mismatch\n got %v\nwant %v", got, wantNames)
|
||||
}
|
||||
|
||||
// Second read, now served from the index, must agree with the first.
|
||||
if got := mustGetMetadata(t, c, "legacy"); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("indexed read disagrees with scan read\n got %v\nwant %v", got, want)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Each single-field write path must add to the index, and each delete path must
|
||||
// remove from it. A field written but not indexed vanishes from every later
|
||||
// read — silently — so this is the property that matters most.
|
||||
func TestFieldIndex_SingleFieldWritePathsStayInSync(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
if err := c.SetProperty("root2", "title", "Naruto"); err != nil {
|
||||
t.Fatalf("SetProperty: %v", err)
|
||||
}
|
||||
if _, err := c.AddToSet("root2", "tags", "anime"); err != nil {
|
||||
t.Fatalf("AddToSet: %v", err)
|
||||
}
|
||||
if _, err := c.MergeMetadataFlat("root2", map[string]string{"year": "2002"}); err != nil {
|
||||
t.Fatalf("MergeMetadataFlat: %v", err)
|
||||
}
|
||||
|
||||
want := map[string]string{"title": "Naruto", "tags": "anime", "year": "2002"}
|
||||
if got := mustGetMetadata(t, c, "root2"); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("after writes\n got %v\nwant %v", got, want)
|
||||
}
|
||||
|
||||
// DeleteProperty drops the key and the index entry.
|
||||
if err := c.DeleteProperty("root2", "title"); err != nil {
|
||||
t.Fatalf("DeleteProperty: %v", err)
|
||||
}
|
||||
// RemoveFromSet deletes the key once its last member is gone.
|
||||
if _, err := c.RemoveFromSet("root2", "tags", "anime"); err != nil {
|
||||
t.Fatalf("RemoveFromSet: %v", err)
|
||||
}
|
||||
|
||||
want = map[string]string{"year": "2002"}
|
||||
if got := mustGetMetadata(t, c, "root2"); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("after deletes\n got %v\nwant %v", got, want)
|
||||
}
|
||||
if got := fieldNames(t, c, "root2"); !reflect.DeepEqual(got, []string{"year"}) {
|
||||
t.Errorf("index after deletes = %v, want [year]", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The raw KV editor can mint or drop a record property out of band. It is the
|
||||
// one write path whose caller does not know the record layout, so the storage
|
||||
// layer has to keep the index honest for it.
|
||||
func TestFieldIndex_RawKeyEditorStaysInSync(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
if err := c.SetRawValue("file:root3/hand-written", "v"); err != nil {
|
||||
t.Fatalf("SetRawValue: %v", err)
|
||||
}
|
||||
if got := fieldNames(t, c, "root3"); !reflect.DeepEqual(got, []string{"hand-written"}) {
|
||||
t.Errorf("index after SetRawValue = %v", got)
|
||||
}
|
||||
if got := mustGetMetadata(t, c, "root3"); !reflect.DeepEqual(got, map[string]string{"hand-written": "v"}) {
|
||||
t.Errorf("read after SetRawValue = %v", got)
|
||||
}
|
||||
|
||||
if _, err := c.DeleteRawKey("file:root3/hand-written"); err != nil {
|
||||
t.Fatalf("DeleteRawKey: %v", err)
|
||||
}
|
||||
if got := fieldNames(t, c, "root3"); len(got) != 0 {
|
||||
t.Errorf("index after DeleteRawKey = %v, want empty", got)
|
||||
}
|
||||
|
||||
// Keys outside the record namespace must be left alone entirely.
|
||||
if err := c.SetRawValue("cid:bafkreiaaa", "root3"); err != nil {
|
||||
t.Fatalf("SetRawValue non-record: %v", err)
|
||||
}
|
||||
if _, _, ok := c.splitRecordKey(c.buildKey("cid:bafkreiaaa")); ok {
|
||||
t.Error("cid: key was treated as a record property")
|
||||
}
|
||||
if _, _, ok := c.splitRecordKey(c.buildIndexKey()); ok {
|
||||
t.Error("file:__index__ was treated as a record property")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Mint and AddAlias write fields through their own pipelines, so they need the
|
||||
// index too. AddAlias in particular is how `cids/<cid>` members appear — miss
|
||||
// it and a record's CID key-set becomes invisible to every read.
|
||||
func TestFieldIndex_MintAndAliasIndexTheirFields(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
uuid, err := c.Mint("/files/watch/x.mkv", 10, 20)
|
||||
if err != nil {
|
||||
t.Fatalf("Mint: %v", err)
|
||||
}
|
||||
if got := fieldNames(t, c, uuid); !reflect.DeepEqual(got, []string{"filePath", "mtimeNano", "sizeByte"}) {
|
||||
t.Errorf("index after Mint = %v", got)
|
||||
}
|
||||
|
||||
if err := c.AddAlias(uuid, "bafkreiaaa"); err != nil {
|
||||
t.Fatalf("AddAlias: %v", err)
|
||||
}
|
||||
got := mustGetMetadata(t, c, uuid)
|
||||
if got["cids/bafkreiaaa"] != "true" {
|
||||
t.Errorf("alias key-set member missing from read: %v", got)
|
||||
}
|
||||
|
||||
// And the CID enumeration, which now reads the index instead of
|
||||
// scanning, must see it.
|
||||
cids, err := c.cidsForRootLocked(context.Background(), uuid)
|
||||
if err != nil {
|
||||
t.Fatalf("cidsForRootLocked: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(cids, []string{"bafkreiaaa"}) {
|
||||
t.Errorf("cidsForRoot = %v, want [bafkreiaaa]", cids)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The index lives under the record's own prefix, so the "is this record empty?"
|
||||
// probe would otherwise always find it and the root could never be de-indexed.
|
||||
func TestFieldIndex_IndexRemoveIfEmptyIgnoresTheIndexItself(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
if err := c.SetProperty("root4", "title", "x"); err != nil {
|
||||
t.Fatalf("SetProperty: %v", err)
|
||||
}
|
||||
|
||||
removed, err := c.IndexRemoveIfEmpty("root4")
|
||||
if err != nil {
|
||||
t.Fatalf("IndexRemoveIfEmpty: %v", err)
|
||||
}
|
||||
if removed {
|
||||
t.Error("de-indexed a record that still has a field")
|
||||
}
|
||||
|
||||
if err := c.DeleteProperty("root4", "title"); err != nil {
|
||||
t.Fatalf("DeleteProperty: %v", err)
|
||||
}
|
||||
removed, err = c.IndexRemoveIfEmpty("root4")
|
||||
if err != nil {
|
||||
t.Fatalf("IndexRemoveIfEmpty: %v", err)
|
||||
}
|
||||
if !removed {
|
||||
t.Fatal("record with no fields left was not de-indexed — the field index masked emptiness")
|
||||
}
|
||||
// ...and the now-empty index key must not be left orphaned.
|
||||
n, err := c.client.Exists(context.Background(), c.buildFieldsKey("root4")).Result()
|
||||
if err != nil {
|
||||
t.Fatalf("exists: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Error("orphaned field index left behind after de-indexing")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The same probe must still work for a record that predates the index.
|
||||
func TestFieldIndex_IndexRemoveIfEmptyHandlesUnindexedRecord(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
ctx := context.Background()
|
||||
if err := c.client.Set(ctx, c.buildKeyPrefix("legacy2")+"title", "x", 0).Err(); err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
if err := c.IndexAdd("legacy2"); err != nil {
|
||||
t.Fatalf("IndexAdd: %v", err)
|
||||
}
|
||||
removed, err := c.IndexRemoveIfEmpty("legacy2")
|
||||
if err != nil {
|
||||
t.Fatalf("IndexRemoveIfEmpty: %v", err)
|
||||
}
|
||||
if removed {
|
||||
t.Error("de-indexed an un-indexed record that still has a field")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The reserved name is bookkeeping. It must never reach an API caller as a
|
||||
// metadata field, and an inbound payload must never be able to overwrite the
|
||||
// SET with a string — that would break every subsequent read of the record.
|
||||
func TestFieldIndex_ReservedNameIsNeverMetadata(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
err := c.SetMetadataFlat("root5", map[string]string{
|
||||
"title": "x",
|
||||
FieldsField: "malicious",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SetMetadataFlat: %v", err)
|
||||
}
|
||||
|
||||
got := mustGetMetadata(t, c, "root5")
|
||||
if _, present := got[FieldsField]; present {
|
||||
t.Errorf("%s surfaced as a metadata field: %v", FieldsField, got)
|
||||
}
|
||||
if !reflect.DeepEqual(got, map[string]string{"title": "x"}) {
|
||||
t.Errorf("read = %v, want just title", got)
|
||||
}
|
||||
|
||||
// The index must still be a SET — i.e. not clobbered by the payload.
|
||||
if got := fieldNames(t, c, "root5"); !reflect.DeepEqual(got, []string{"title"}) {
|
||||
t.Errorf("index = %v, want [title]", got)
|
||||
}
|
||||
|
||||
if err := c.SetProperty("root5", FieldsField, "malicious"); err == nil {
|
||||
t.Error("SetProperty accepted the reserved field name")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Deleting a record must take its index with it, or a recreated root would
|
||||
// inherit phantom field names.
|
||||
func TestFieldIndex_DeleteMetadataRemovesTheIndex(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
if err := c.SetMetadataFlat("root6", map[string]string{"title": "x", "year": "2002"}); err != nil {
|
||||
t.Fatalf("SetMetadataFlat: %v", err)
|
||||
}
|
||||
if _, err := c.DeleteMetadata("root6"); err != nil {
|
||||
t.Fatalf("DeleteMetadata: %v", err)
|
||||
}
|
||||
if got := fieldNames(t, c, "root6"); len(got) != 0 {
|
||||
t.Errorf("index survived DeleteMetadata: %v", got)
|
||||
}
|
||||
got, err := c.GetMetadataFlat("root6")
|
||||
if err != nil {
|
||||
t.Fatalf("GetMetadataFlat: %v", err)
|
||||
}
|
||||
if got != nil {
|
||||
t.Errorf("deleted record still reads back: %v", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A stale index entry (name present, key gone) is the benign direction: the
|
||||
// MGET reads nil and the field is skipped rather than erroring.
|
||||
func TestFieldIndex_StaleEntryIsHarmless(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
ctx := context.Background()
|
||||
if err := c.SetMetadataFlat("root7", map[string]string{"title": "x"}); err != nil {
|
||||
t.Fatalf("SetMetadataFlat: %v", err)
|
||||
}
|
||||
// Name a field that has no key behind it.
|
||||
if err := c.client.SAdd(ctx, c.buildFieldsKey("root7"), "ghost").Err(); err != nil {
|
||||
t.Fatalf("sadd ghost: %v", err)
|
||||
}
|
||||
|
||||
got := mustGetMetadata(t, c, "root7")
|
||||
if !reflect.DeepEqual(got, map[string]string{"title": "x"}) {
|
||||
t.Errorf("read = %v, want just title", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// WarmFieldIndexes is what keeps an upgrade from spending hours reading
|
||||
// un-indexed records one full-keyspace scan at a time. It must index every
|
||||
// record in one pass, be idempotent, and leave already-indexed records alone.
|
||||
func TestFieldIndex_WarmBuildsEveryMissingIndex(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
ctx := context.Background()
|
||||
|
||||
// Two legacy records seeded straight into Redis (no index), plus one
|
||||
// written through the normal path (already indexed).
|
||||
legacy := map[string]map[string]string{
|
||||
"old1": {"filePath": "/a.mkv", "cids/bafkreiaaa": "true"},
|
||||
"old2": {"filePath": "/b.mkv", "titles/jpn": "ナルト", "deep/a/b": "x"},
|
||||
}
|
||||
for root, fields := range legacy {
|
||||
for f, v := range fields {
|
||||
if err := c.client.Set(ctx, c.buildKeyPrefix(root)+f, v, 0).Err(); err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
}
|
||||
if err := c.IndexAdd(root); err != nil {
|
||||
t.Fatalf("IndexAdd: %v", err)
|
||||
}
|
||||
}
|
||||
if err := c.SetMetadataFlat("fresh", map[string]string{"title": "x"}); err != nil {
|
||||
t.Fatalf("SetMetadataFlat: %v", err)
|
||||
}
|
||||
|
||||
n, err := c.WarmFieldIndexes()
|
||||
if err != nil {
|
||||
t.Fatalf("WarmFieldIndexes: %v", err)
|
||||
}
|
||||
if n < 2 {
|
||||
t.Errorf("warmed %d records, want at least the 2 legacy ones", n)
|
||||
}
|
||||
|
||||
if got := fieldNames(t, c, "old1"); !reflect.DeepEqual(got, []string{"cids/bafkreiaaa", "filePath"}) {
|
||||
t.Errorf("old1 index = %v", got)
|
||||
}
|
||||
if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) {
|
||||
t.Errorf("old2 index = %v", got)
|
||||
}
|
||||
if got := fieldNames(t, c, "fresh"); !reflect.DeepEqual(got, []string{"title"}) {
|
||||
t.Errorf("fresh index disturbed = %v", got)
|
||||
}
|
||||
|
||||
// Reads must now agree with what the records actually hold.
|
||||
for root, want := range legacy {
|
||||
if got := mustGetMetadata(t, c, root); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("%s read = %v, want %v", root, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Idempotent: a second pass changes nothing.
|
||||
if _, err := c.WarmFieldIndexes(); err != nil {
|
||||
t.Fatalf("second WarmFieldIndexes: %v", err)
|
||||
}
|
||||
if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) {
|
||||
t.Errorf("old2 index after second warm = %v", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The bulk search read enumerates the whole file: keyspace and MGETs what it
|
||||
// finds. The index keys are SETs living in that same namespace, so it has to
|
||||
// skip them or the MGET fails with WRONGTYPE and search dies wholesale.
|
||||
func TestFieldIndex_SearchIndexTuplesSkipTheIndexKeys(t *testing.T) {
|
||||
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||
if err := c.SetMetadataFlat("root8", map[string]string{
|
||||
"title": "Naruto Shippuden",
|
||||
"fileName": "naruto.mkv",
|
||||
}); err != nil {
|
||||
t.Fatalf("SetMetadataFlat: %v", err)
|
||||
}
|
||||
|
||||
tuples, err := c.GetSearchIndexTuples()
|
||||
if err != nil {
|
||||
t.Fatalf("GetSearchIndexTuples: %v", err)
|
||||
}
|
||||
var found bool
|
||||
for _, tu := range tuples {
|
||||
if tu.HashID != "root8" {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if _, present := tu.Fields[FieldsField]; present {
|
||||
t.Errorf("%s leaked into the search tuple: %v", FieldsField, tu.Fields)
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("record missing from search tuples")
|
||||
}
|
||||
})
|
||||
}
|
||||
+22
-4
@@ -72,7 +72,15 @@ func udlIdxCidUsers(cid string) string { return fmt.Sprintf("udl:idx:cid:%s
|
||||
//
|
||||
// KEYS: 1=cell 2=idx:user:key 3=idx:user:cid 4=idx:cid:key 5=idx:cid:users
|
||||
// ARGV: 1=version 2=ts 3=record 4=value 5=hasValue("1"/"0")
|
||||
// 6=cid 7=key 8=uid
|
||||
// 6=cid 7=key 8=uid 9=tombstone("1"/"0")
|
||||
//
|
||||
// A tombstone still writes the cell — that is the CRDT state, and dropping it
|
||||
// would let an older write resurrect the value — but it SREMs the cid from the
|
||||
// user+key index. That index is what `active_for_user_key` returns, so leaving
|
||||
// deleted cids in it means every My List / Continue Watching read carries every
|
||||
// title ever un-liked or finished, forever. The other three indexes are left
|
||||
// alone: they answer "who has an opinion about this cid", where a tombstone is
|
||||
// still an answer.
|
||||
var udlUpsertScript = redis.NewScript(`
|
||||
local stored = redis.call('HGET', KEYS[1], 'version')
|
||||
if stored and tonumber(stored) >= tonumber(ARGV[1]) then
|
||||
@@ -84,7 +92,11 @@ if ARGV[5] == '1' then
|
||||
else
|
||||
redis.call('HDEL', KEYS[1], 'value')
|
||||
end
|
||||
if ARGV[9] == '1' then
|
||||
redis.call('SREM', KEYS[2], ARGV[6])
|
||||
else
|
||||
redis.call('SADD', KEYS[2], ARGV[6])
|
||||
end
|
||||
redis.call('SADD', KEYS[3], ARGV[7])
|
||||
redis.call('SADD', KEYS[4], ARGV[8])
|
||||
redis.call('SADD', KEYS[5], ARGV[8])
|
||||
@@ -94,8 +106,10 @@ return 1
|
||||
// UDLUpsertIfNewer stores a record cell iff version strictly exceeds the stored
|
||||
// one. publicValue/hasValue carry the plaintext JSON scalar for public keys
|
||||
// (used by aggregation); pass hasValue=false for private-tier keys and
|
||||
// tombstones. Returns accepted=false when the write is stale.
|
||||
func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, record, publicValue string, hasValue bool) (bool, error) {
|
||||
// tombstones. Set tombstone=true when the record deletes the cell, which also
|
||||
// prunes the cid from the user+key index. Returns accepted=false when the write
|
||||
// is stale.
|
||||
func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, record, publicValue string, hasValue, tombstone bool) (bool, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
@@ -117,7 +131,11 @@ func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, recor
|
||||
if hasValue {
|
||||
hv = "1"
|
||||
}
|
||||
res, err := udlUpsertScript.Run(ctx, c.client, keys, version, ts, record, publicValue, hv, cid, key, uid).Result()
|
||||
tomb := "0"
|
||||
if tombstone {
|
||||
tomb = "1"
|
||||
}
|
||||
res, err := udlUpsertScript.Run(ctx, c.client, keys, version, ts, record, publicValue, hv, cid, key, uid, tomb).Result()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("udl upsert failed: %w", err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A tombstone must leave the cell in place but drop the cid from the user+key
|
||||
// index.
|
||||
//
|
||||
// The cell has to stay: it *is* the CRDT state, and deleting it would let an
|
||||
// older write resurrect the value the tombstone retracted. The index entry has
|
||||
// to go: `active_for_user_key` returns every member of that set, so a set that
|
||||
// only ever grows means each My List / Continue Watching read carries every
|
||||
// title the user has ever un-liked or finished — with its full record blob —
|
||||
// forever. The reader filters them out again, so the cost is invisible until
|
||||
// the account is old.
|
||||
func TestUDLUpsert_TombstonePrunesUserKeyIndexButKeepsCell(t *testing.T) {
|
||||
c, _ := newTestClient(t, "")
|
||||
|
||||
const (
|
||||
uid = "zAlice"
|
||||
cid = "bagcsaaalocator"
|
||||
key = "seek"
|
||||
)
|
||||
|
||||
// A live position: indexed, readable.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-v1", "", false, false); err != nil || !ok {
|
||||
t.Fatalf("live upsert: ok=%v err=%v", ok, err)
|
||||
}
|
||||
entries, err := c.UDLListUserKey(uid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("active_for_user_key: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].Cid != cid {
|
||||
t.Fatalf("live position should be indexed, got %+v", entries)
|
||||
}
|
||||
|
||||
// Finishing it tombstones the seek.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 2, 200, "rec-v2-tombstone", "", false, true); err != nil || !ok {
|
||||
t.Fatalf("tombstone upsert: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
entries, err = c.UDLListUserKey(uid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("active_for_user_key after tombstone: %v", err)
|
||||
}
|
||||
if len(entries) != 0 {
|
||||
t.Fatalf("tombstoned cid must leave the user+key index, got %+v", entries)
|
||||
}
|
||||
|
||||
// ...but the cell itself survives, at the tombstone's version.
|
||||
cell, exists, err := c.UDLGetCell(uid, cid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("get cell: %v", err)
|
||||
}
|
||||
if !exists {
|
||||
t.Fatal("tombstone must not delete the cell — an older write could then resurrect the value")
|
||||
}
|
||||
if cell.Version != 2 || cell.Record != "rec-v2-tombstone" {
|
||||
t.Fatalf("cell should hold the tombstone record at v2, got %+v", cell)
|
||||
}
|
||||
|
||||
// And the high-water mark still rejects the superseded write.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 300, "rec-v1-replay", "", false, false); err != nil || ok {
|
||||
t.Fatalf("stale replay must be rejected: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
// Re-watching re-establishes the cell and puts it back in the index.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 3, 400, "rec-v3", "", false, false); err != nil || !ok {
|
||||
t.Fatalf("re-establish upsert: ok=%v err=%v", ok, err)
|
||||
}
|
||||
entries, err = c.UDLListUserKey(uid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("active_for_user_key after re-establish: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].Cid != cid {
|
||||
t.Fatalf("re-established cid must return to the index, got %+v", entries)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user