#225: read-trail master switch and purpose limitation #281
@ -0,0 +1,32 @@
|
||||
-- #223 (ADR 0023): dedup window for the read trail. Aligned buckets
|
||||
-- (floor(epoch / window)) with a unique (dedup_key, window_bucket) pair make
|
||||
-- concurrent duplicates collapse race-free at insert time.
|
||||
ALTER TABLE "read_events"
|
||||
ADD COLUMN "dedup_key" TEXT,
|
||||
ADD COLUMN "window_bucket" BIGINT,
|
||||
ADD COLUMN "window_seconds" INTEGER;
|
||||
|
||||
-- Backfill rows written between the #222 and #223 deploys under the default
|
||||
-- 5-minute window, then apply the window's own semantics retroactively:
|
||||
-- within one (key, bucket) pair only the FIRST event is the evidence row —
|
||||
-- exactly what the window would have recorded had it existed.
|
||||
UPDATE "read_events"
|
||||
SET "dedup_key" = "session_key" || ':' || COALESCE("page_id", '-') || ':' || "channel",
|
||||
"window_bucket" = FLOOR(EXTRACT(EPOCH FROM "occurred_at") / 300)::BIGINT,
|
||||
"window_seconds" = 300
|
||||
WHERE "dedup_key" IS NULL;
|
||||
|
||||
DELETE FROM "read_events" keep
|
||||
USING "read_events" first
|
||||
WHERE keep."dedup_key" = first."dedup_key"
|
||||
AND keep."window_bucket" = first."window_bucket"
|
||||
AND (first."occurred_at" < keep."occurred_at"
|
||||
OR (first."occurred_at" = keep."occurred_at" AND first."id" < keep."id"));
|
||||
|
||||
ALTER TABLE "read_events"
|
||||
ALTER COLUMN "dedup_key" SET NOT NULL,
|
||||
ALTER COLUMN "window_bucket" SET NOT NULL,
|
||||
ALTER COLUMN "window_seconds" SET NOT NULL;
|
||||
|
||||
CREATE UNIQUE INDEX "read_events_dedup_key_window_bucket_key"
|
||||
ON "read_events"("dedup_key", "window_bucket");
|
||||
@ -115,7 +115,16 @@ model ReadEvent {
|
||||
/// rewrite history (ADR 0023).
|
||||
classification String
|
||||
details Json?
|
||||
/// Dedup window (issue #223): `<sessionKey>:<pageId|->:<channel>` plus the
|
||||
/// aligned bucket `floor(epoch / windowSeconds)`. The unique pair makes
|
||||
/// concurrent duplicate reads collapse race-free (insert or P2002-skip).
|
||||
dedupKey String @map("dedup_key")
|
||||
windowBucket BigInt @map("window_bucket")
|
||||
/// Window length the event was recorded under — the row itself states it
|
||||
/// represents up to this many seconds, so the evidence is not overread.
|
||||
windowSeconds Int @map("window_seconds")
|
||||
|
||||
@@unique([dedupKey, windowBucket])
|
||||
@@index([pageId, occurredAt])
|
||||
@@index([actorId, occurredAt])
|
||||
@@index([occurredAt])
|
||||
|
||||
177
apps/api/src/read-trail/read-trail-dedup.e2e.db.test.ts
Normal file
177
apps/api/src/read-trail/read-trail-dedup.e2e.db.test.ts
Normal file
@ -0,0 +1,177 @@
|
||||
import { INestApplication } from '@nestjs/common';
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import request from 'supertest';
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { ClockService } from '../common/clock.service';
|
||||
import { createTestApp, sessionCookieOf } from '../testing/test-app';
|
||||
import { createTestPrisma, grantOwnerAdmin, hasTestDb, uniqueSuffix } from '../testing/test-db';
|
||||
import { UsersService } from '../users/users.service';
|
||||
|
||||
const PNG_SIGNATURE = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);
|
||||
|
||||
/**
|
||||
* The read-trail dedup window (issue #223, ADR 0023): one event per
|
||||
* (session, page, channel) within an aligned `readTrail.dedupWindowMinutes`
|
||||
* window — repeats and reconnect-style re-requests collapse, a new session
|
||||
* or a new window does not, and the recorded row states the window length
|
||||
* it represents.
|
||||
*/
|
||||
describe.skipIf(!hasTestDb)('read-trail dedup window (e2e, issue #223)', () => {
|
||||
let app: INestApplication;
|
||||
let prisma: PrismaClient;
|
||||
const suffix = uniqueSuffix();
|
||||
const password = 'dedup fenster zeugen 123';
|
||||
|
||||
let ownerId: string;
|
||||
let ownerCookie: string;
|
||||
let pondId: string;
|
||||
let classifiedId: string;
|
||||
|
||||
const api = () => request(app.getHttpServer());
|
||||
const login = async () =>
|
||||
sessionCookieOf(
|
||||
await api()
|
||||
.post('/api/v1/auth/login')
|
||||
.send({ usernameOrEmail: `dedup-owner-${suffix}`, password })
|
||||
.expect(200),
|
||||
);
|
||||
const events = () =>
|
||||
prisma.readEvent.findMany({ where: { pondId }, orderBy: { occurredAt: 'asc' } });
|
||||
|
||||
beforeAll(async () => {
|
||||
prisma = createTestPrisma();
|
||||
await prisma.rateLimit.deleteMany({});
|
||||
app = await createTestApp();
|
||||
const users = app.get(UsersService);
|
||||
|
||||
const owner = await users.createUser({
|
||||
username: `dedup-owner-${suffix}`,
|
||||
email: `dedup-owner-${suffix}@example.test`,
|
||||
displayName: 'Dedup Owner',
|
||||
password,
|
||||
locale: 'en',
|
||||
});
|
||||
await users.markEmailVerified(owner.id);
|
||||
ownerId = owner.id;
|
||||
ownerCookie = await login();
|
||||
|
||||
const pond = await prisma.pond.create({
|
||||
data: { slug: `dedup-pond-${suffix}`, name: 'Dedup Pond', type: 'SHARED', ownerId },
|
||||
});
|
||||
pondId = pond.id;
|
||||
await grantOwnerAdmin(prisma, pondId, ownerId);
|
||||
|
||||
const page = await prisma.page.create({
|
||||
data: {
|
||||
pondId,
|
||||
slug: `classified-${suffix}`,
|
||||
title: 'Classified',
|
||||
classification: 'VS_NFD',
|
||||
createdBy: ownerId,
|
||||
sortKey: 'a0',
|
||||
ydocState: new Uint8Array(),
|
||||
contentCache: {
|
||||
create: { plainText: 'x', markdown: 'x', html: '<p>x</p>', outline: [] },
|
||||
},
|
||||
},
|
||||
});
|
||||
classifiedId = page.id;
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await prisma.readEvent.deleteMany({ where: { pondId } });
|
||||
await prisma.rateLimit.deleteMany({});
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await prisma.readEvent.deleteMany({ where: { pondId } });
|
||||
await prisma.attachment.deleteMany({ where: { pondId } });
|
||||
await prisma.roleGrant.deleteMany({ where: { pondId } });
|
||||
await prisma.pageContentCache.deleteMany({ where: { page: { pondId } } });
|
||||
await prisma.page.deleteMany({ where: { pondId } });
|
||||
await prisma.pond.deleteMany({ where: { id: pondId } });
|
||||
await prisma.user.deleteMany({ where: { id: ownerId } });
|
||||
await prisma.$disconnect();
|
||||
await app.close();
|
||||
});
|
||||
|
||||
it('collapses repeated reads of one page in one session+channel to a single event that names its window', async () => {
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
const rows = await events();
|
||||
expect(rows).toHaveLength(1);
|
||||
// The row itself states that it represents a window, not a request.
|
||||
expect(rows[0]!.windowSeconds).toBe(5 * 60);
|
||||
expect(rows[0]!.dedupKey).toBe(`${rows[0]!.sessionKey}:${classifiedId}:page_view`);
|
||||
});
|
||||
|
||||
it('keeps channels apart: the same session reading and joining collab yields one event each', async () => {
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
await api()
|
||||
.get(`/api/v1/pages/${classifiedId}/collab-token`)
|
||||
.set('Cookie', ownerCookie)
|
||||
.expect(200);
|
||||
const rows = await events();
|
||||
expect(rows.map((r) => r.channel).sort()).toEqual(['collab_join', 'page_view']);
|
||||
});
|
||||
|
||||
it('a new session records again even for the same user — a reconnect within the session does not', async () => {
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
// Same session, later request ("reconnect"): deduped.
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
expect(await events()).toHaveLength(1);
|
||||
|
||||
const secondCookie = await login();
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', secondCookie).expect(200);
|
||||
const rows = await events();
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(new Set(rows.map((r) => r.sessionKey)).size).toBe(2);
|
||||
expect(rows.every((r) => r.actorId === ownerId)).toBe(true);
|
||||
});
|
||||
|
||||
it('bounds a live editing session: 30 collab-token renewals in one window are one event', async () => {
|
||||
for (let i = 0; i < 30; i += 1) {
|
||||
await api()
|
||||
.get(`/api/v1/pages/${classifiedId}/collab-token`)
|
||||
.set('Cookie', ownerCookie)
|
||||
.expect(200);
|
||||
}
|
||||
const rows = await events();
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0]!.channel).toBe('collab_join');
|
||||
});
|
||||
|
||||
it('opens a new window when the clock moves past the bucket boundary', async () => {
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
const clock = app.get(ClockService);
|
||||
const later = new Date(Date.now() + 10 * 60 * 1000);
|
||||
const spy = vi.spyOn(clock, 'now').mockReturnValue(later);
|
||||
try {
|
||||
await api().get(`/api/v1/pages/${classifiedId}`).set('Cookie', ownerCookie).expect(200);
|
||||
} finally {
|
||||
spy.mockRestore();
|
||||
}
|
||||
expect(await events()).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('dedups page-less attachment downloads on the placeholder page key', async () => {
|
||||
// A pond-level upload has no page; its effective classification falls
|
||||
// back to the pond maximum (#212) and the dedup key carries `-`.
|
||||
const uploaded = await api()
|
||||
.post(`/api/v1/ponds/${pondId}/files`)
|
||||
.set('Cookie', ownerCookie)
|
||||
.attach('file', Buffer.concat([PNG_SIGNATURE, Buffer.from('img')]), 'a.png')
|
||||
.expect(201);
|
||||
await prisma.readEvent.deleteMany({ where: { pondId } });
|
||||
|
||||
await api().get(`/api/v1/media/${uploaded.body.id}`).set('Cookie', ownerCookie).expect(200);
|
||||
await api().get(`/api/v1/media/${uploaded.body.id}`).set('Cookie', ownerCookie).expect(200);
|
||||
const rows = await events();
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0]!.channel).toBe('attachment');
|
||||
expect(rows[0]!.pageId).toBeNull();
|
||||
expect(rows[0]!.dedupKey).toBe(`${rows[0]!.sessionKey}:-:attachment`);
|
||||
});
|
||||
});
|
||||
@ -1,5 +1,8 @@
|
||||
import { Global, Module } from '@nestjs/common';
|
||||
|
||||
import { CommonModule } from '../common/common.module';
|
||||
import { SettingsModule } from '../settings/settings.module';
|
||||
|
||||
import { ReadTrailService } from './read-trail.service';
|
||||
|
||||
/**
|
||||
@ -9,6 +12,7 @@ import { ReadTrailService } from './read-trail.service';
|
||||
*/
|
||||
@Global()
|
||||
@Module({
|
||||
imports: [CommonModule, SettingsModule],
|
||||
providers: [ReadTrailService],
|
||||
exports: [ReadTrailService],
|
||||
})
|
||||
|
||||
@ -2,7 +2,9 @@ import { Injectable } from '@nestjs/common';
|
||||
import { Prisma } from '@prisma/client';
|
||||
import { PinoLogger } from 'nestjs-pino';
|
||||
|
||||
import { ClockService } from '../common/clock.service';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { InstanceSettingsService } from '../settings/instance-settings.service';
|
||||
|
||||
/**
|
||||
* Every read surface classified content can leave through (issue #222,
|
||||
@ -61,11 +63,22 @@ export interface ReadEventInput extends ReadActor {
|
||||
* swallowed. A lost event is a gap in evidence, so a failed write aborts
|
||||
* the read with the ordinary 500 — the reader retries, the evidence stays
|
||||
* complete (decision recorded in ADR 0023 and security.md).
|
||||
*
|
||||
* Dedup window (issue #223): one event per (session, page, channel) within
|
||||
* an aligned window of `readTrail.dedupWindowMinutes` (default 5). Buckets
|
||||
* are `floor(epoch / windowSeconds)`, and the unique
|
||||
* (dedupKey, windowBucket) pair collapses concurrent duplicates race-free:
|
||||
* the first insert wins, every later one lands on P2002 and is skipped —
|
||||
* a skipped DUPLICATE is not a gap, so it must not abort the read. The row
|
||||
* carries `windowSeconds`, so each event states that it represents up to
|
||||
* that many seconds of access, not a single request.
|
||||
*/
|
||||
@Injectable()
|
||||
export class ReadTrailService {
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly settings: InstanceSettingsService,
|
||||
private readonly clock: ClockService,
|
||||
private readonly logger: PinoLogger,
|
||||
) {
|
||||
this.logger.setContext(ReadTrailService.name);
|
||||
@ -73,21 +86,35 @@ export class ReadTrailService {
|
||||
|
||||
async record(event: ReadEventInput): Promise<void> {
|
||||
const { actorId, sessionKey, pageId, pondId, channel, details } = event;
|
||||
await this.prisma.readEvent.create({
|
||||
data: {
|
||||
actorId,
|
||||
sessionKey,
|
||||
pageId,
|
||||
pondId,
|
||||
channel,
|
||||
classification: 'vs_nfd',
|
||||
details: details ? (details as Prisma.InputJsonObject) : undefined,
|
||||
},
|
||||
});
|
||||
const windowSeconds = (await this.settings.get('readTrail.dedupWindowMinutes')) * 60;
|
||||
const dedupKey = `${sessionKey}:${pageId ?? '-'}:${channel}`;
|
||||
const windowBucket = BigInt(Math.floor(this.clock.now().getTime() / 1000 / windowSeconds));
|
||||
try {
|
||||
await this.prisma.readEvent.create({
|
||||
data: {
|
||||
actorId,
|
||||
sessionKey,
|
||||
pageId,
|
||||
pondId,
|
||||
channel,
|
||||
classification: 'vs_nfd',
|
||||
details: details ? (details as Prisma.InputJsonObject) : undefined,
|
||||
dedupKey,
|
||||
windowBucket,
|
||||
windowSeconds,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
const isDuplicate =
|
||||
error instanceof Prisma.PrismaClientKnownRequestError && error.code === 'P2002';
|
||||
if (!isDuplicate) throw error;
|
||||
this.logger.debug({ dedupKey, channel }, 'read_trail: deduped within window');
|
||||
return;
|
||||
}
|
||||
// The stdout line mirrors the row (SIEM forwarding beyond this is out of
|
||||
// scope, #224); it fires only after the row is safely persisted.
|
||||
this.logger.info(
|
||||
{ actor: actorId, sessionKey, pageId, pondId, channel },
|
||||
{ actor: actorId, sessionKey, pageId, pondId, channel, windowSeconds },
|
||||
'read_trail: classified page read',
|
||||
);
|
||||
}
|
||||
|
||||
@ -51,6 +51,11 @@ export const INSTANCE_SETTINGS = {
|
||||
// bounded. PENDING rows — including failed-but-retryable ones — are
|
||||
// never touched; the retry loop owns them.
|
||||
'mail.outboxRetentionDays': z.number().int().min(1).default(30),
|
||||
// Read-trail dedup window (issue #223, ADR 0023): one event per
|
||||
// (session, page, channel) within an aligned window of this many minutes.
|
||||
// 5 minutes keeps a live Yjs session (collab tokens every 60 s) at a
|
||||
// bounded ~12 events/hour/page while still evidencing distinct visits.
|
||||
'readTrail.dedupWindowMinutes': z.number().int().min(1).default(5),
|
||||
// Default VS-NfD classification for newly created pages (ADR 0022,
|
||||
// issue #204). An instance operated inside a classified environment sets
|
||||
// this to `vs_nfd` so nothing starts unmarked; inheritance from the
|
||||
|
||||
@ -67,6 +67,28 @@ traffic per open document.
|
||||
also the MCP `read_page` path), `job:<id>` (background builds such as the
|
||||
account data export), `anon` (anonymous reader on a public grant).
|
||||
|
||||
## Decisions taken in #223
|
||||
|
||||
- **Aligned dedup windows.** One event per (session, page, channel) within
|
||||
an aligned window of `readTrail.dedupWindowMinutes` (default 5):
|
||||
buckets are `floor(epoch / windowSeconds)`, and a unique
|
||||
(dedupKey, windowBucket) pair collapses concurrent duplicates race-free
|
||||
at insert time. Sliding windows (measured from the first event) were
|
||||
rejected — they need a read-before-write and lose the race-safety of the
|
||||
plain unique insert. Consequence: two reads just across a bucket
|
||||
boundary yield two events; over-recording is acceptable, gaps are not.
|
||||
- **A skipped duplicate is not a gap** — the unique-violation path returns
|
||||
quietly (debug log), only real write failures abort the read.
|
||||
- **Each row states its window** (`windowSeconds`), so the evidence reads
|
||||
as "accessed at least once in these N minutes", never as a request
|
||||
count. What the trail can prove about a live session: the per-minute
|
||||
collab-token renewals collapse to one `collab_join` event per window —
|
||||
presence during the window, not activity within it.
|
||||
- **Anonymous readers share one `anon` key**: all anonymous reads of a page
|
||||
through one channel inside a window are one event. Deliberate — telling
|
||||
anonymous browsers apart would require fingerprinting (IP/UA), which the
|
||||
purpose limitation (#225) rules out.
|
||||
|
||||
## Consequences
|
||||
|
||||
- The scope limit is the feature's strongest argument in the works-council
|
||||
|
||||
@ -134,7 +134,7 @@ Protokolliert werden Lesezugriffe **ausschließlich** auf Seiten mit
|
||||
|
||||
- [x] Instrumentierung der Lesepfade: Seitenansicht, Public-API-GET,
|
||||
Attachment-Download, Export, No-JS-Shell, Collab-WS-Join · 4 AT · #222
|
||||
- [ ] Dedup-Fenster (eine Sitzung + eine Seite innerhalb N Minuten = ein
|
||||
- [x] Dedup-Fenster (eine Sitzung + eine Seite innerhalb N Minuten = ein
|
||||
Ereignis), sonst erzeugt Yjs-Sync eine Ereignisflut · 2 AT · #223
|
||||
- [ ] Getrennte Tabelle mit eigener Retention und Partitionierung · 2 AT · #224
|
||||
- [ ] Abschaltbar, Zweckbindung dokumentiert · 1–2 AT · #225
|
||||
|
||||
@ -39,6 +39,7 @@ Settings-Cache ist in-process (operations.md).
|
||||
| `upload.allowedExtensions` | nur das dienstlich Nötige (z. B. `pdf`) | Standardliste | jede zusätzliche Endung vergrößert die Menge nicht prüfbarer Binärformate im Bestand. Bilder sind davon unabhängig immer erlaubt (Magic-Byte-geprüft). |
|
||||
| `backup.nextcloud.enabled` | `false` | `false` | „Backup nur lokal": kein Anwendungs-Upload von Restore-Sets zu Drittdiensten. Fernspiegel regelt ausschließlich die Deploy-Allowlist (1.2). |
|
||||
| `trash.retentionDays`, `audit.retentionDays`, `conversion.payloadRetentionDays`, `mail.outboxRetentionDays` | Defaults (30/365/30/30) | ebd. | Aufbewahrung bewusst begrenzt; Verkürzung nach Betreiber-Löschkonzept zulässig (Betriebshandbuch §5). |
|
||||
| `readTrail.dedupWindowMinutes` | Default (5) | `5` | Dedup-Fenster des Lesetrails (#223): je (Sitzung, Seite, Kanal) ein Ereignis pro Fenster — begrenzt die Ereignisflut einer Live-Sitzung auf ~12/h. Kleiner = feineres Protokoll und mehr Zeilen; Änderung mit dem Zweckbindungs-Dokument (#225) abstimmen. |
|
||||
| `legal.imprint`, `legal.privacyPolicy` | befüllt | leer | Betreiberpflicht; leere Seiten zeigen einen Warnbanner. |
|
||||
|
||||
### 1.2 Deploy-Konfiguration (`.env` / Compose — nur Plattformzugriff, bewusst nicht per Admin-UI)
|
||||
|
||||
@ -181,7 +181,14 @@ Datenträger- und Transportschutz ist Plattformsache
|
||||
### 3.5 Lesekanäle
|
||||
|
||||
Jeder Kanal, über den Seiteninhalt die Anwendung verlässt (zugleich die
|
||||
Instrumentierungsliste für den Lesetrail, #222):
|
||||
Instrumentierungsliste für den Lesetrail, #222). Der Lesetrail
|
||||
protokolliert je (Sitzung, Seite, Kanal) **ein Ereignis pro
|
||||
Dedup-Fenster** (`readTrail.dedupWindowMinutes`, Default 5 Minuten,
|
||||
#223); jede Zeile trägt die Fensterlänge. Beweiswert damit: „mindestens
|
||||
ein Zugriff in diesem Fenster" — **nicht** eine Zugriffszählung, und
|
||||
beim Kanal Collab „Live-Verbindung bestand während des Fensters", nicht
|
||||
einzelne Sync-Frames. Anonyme Leser teilen sich den Marker `anon`
|
||||
(keine Fingerprinting-Unterscheidung, Zweckbindung #225):
|
||||
|
||||
| Kanal | Pfad | Rechteprüfung | Kennzeichnung (M26) |
|
||||
| -------------------------- | -------------------------------- | -------------------------------------- | ----------------------------------- |
|
||||
|
||||
Loading…
Reference in New Issue
Block a user