rickub
nandi/gleanpublic⑂ Fork 0
⑂ 0e0f3d1
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/glean.git
git clone ssh://git@rickub.com/nandi/glean.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Stop ingesting the whole network so the database can stay writable

Jetstream was storing every follow/like/annotation on the firehose, which
filled the 5GB volume and made OAuth callback writes fail. Scope the
subscription to known users, purge rows that do not belong to them, and
drop articles older than 30 days so sign-in can persist a session again.
nandi committed 2026-08-22T23:03:34-07:00 Browse files
0e0f3d1 parent: 49597d1
modified .env.example +2 -0
@@ -19,6 +19,8 @@ GLEAN_PLC_URL=https://plc.eurosky.network
1919 GLEAN_SYNC_INTERVAL=8h
2020 GLEAN_CLUSTER_INTERVAL=60m
2121 GLEAN_FETCH_INTERVAL=15m
22+# Delete articles older than this many days (by published date). 0 disables.
23+GLEAN_ARTICLE_RETENTION_DAYS=30
2224 GLEAN_COLLECTION_DIR_URL=https://relay1.us-west.bsky.network/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription
2325 GLEAN_BACKFILL_CONCURRENCY=5
2426
@@ -19,6 +19,8 @@ GLEAN_PLC_URL=https://plc.eurosky.network
19 GLEAN_SYNC_INTERVAL=8h19 GLEAN_SYNC_INTERVAL=8h
20 GLEAN_CLUSTER_INTERVAL=60m20 GLEAN_CLUSTER_INTERVAL=60m
21 GLEAN_FETCH_INTERVAL=15m21 GLEAN_FETCH_INTERVAL=15m
22+# Delete articles older than this many days (by published date). 0 disables.
23+GLEAN_ARTICLE_RETENTION_DAYS=30
22 GLEAN_COLLECTION_DIR_URL=https://relay1.us-west.bsky.network/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription24 GLEAN_COLLECTION_DIR_URL=https://relay1.us-west.bsky.network/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription
23 GLEAN_BACKFILL_CONCURRENCY=525 GLEAN_BACKFILL_CONCURRENCY=5
24 26
modified docs/specs.md +5 -0
@@ -299,6 +299,11 @@ Glean subscribes to a Jetstream endpoint (`GLEAN_JETSTREAM`, default `wss://jets
299299 SUBSCRIBE collections: ["at.glean.subscription", "at.glean.annotation", "at.glean.like", "app.bsky.graph.follow", "sh.tangled.graph.follow", "at.margin.note", "app.skyreader.feed.subscription"]
300300 ```
301301
302+The subscription is scoped with `wantedDids` to the DIDs of known users
303+(the `users` table): events of the wider network are never delivered or
304+stored. The filter is refreshed on every reconnect, and connections are
305+rotated periodically so newly signed-up users start streaming.
306+
302307 On each event:
303308
304309 - **create**: Insert record into local SQLite, update materialized counts
@@ -299,6 +299,11 @@ Glean subscribes to a Jetstream endpoint (`GLEAN_JETSTREAM`, default `wss://jets
299 SUBSCRIBE collections: ["at.glean.subscription", "at.glean.annotation", "at.glean.like", "app.bsky.graph.follow", "sh.tangled.graph.follow", "at.margin.note", "app.skyreader.feed.subscription"]299 SUBSCRIBE collections: ["at.glean.subscription", "at.glean.annotation", "at.glean.like", "app.bsky.graph.follow", "sh.tangled.graph.follow", "at.margin.note", "app.skyreader.feed.subscription"]
300 ```300 ```
301 301
302+The subscription is scoped with `wantedDids` to the DIDs of known users
303+(the `users` table): events of the wider network are never delivered or
304+stored. The filter is refreshed on every reconnect, and connections are
305+rotated periodically so newly signed-up users start streaming.
306+
302 On each event:307 On each event:
303 308
304 - **create**: Insert record into local SQLite, update materialized counts309 - **create**: Insert record into local SQLite, update materialized counts
modified internal/atproto/jetstream.go +61 -12
@@ -76,15 +76,29 @@ func (s *jetstreamScheduler) AddWork(ctx context.Context, _ string, evt *models.
7676
7777 func (s *jetstreamScheduler) Shutdown() {}
7878
79+// KnownDIDsProvider returns the DIDs of all users Glean currently knows.
80+type KnownDIDsProvider func(ctx context.Context) ([]string, error)
81+
82+const (
83+ defaultRewind = 5 * time.Second
84+ // Connections are rotated regularly so an updated known-user filter takes
85+ // effect: wantedDids is fixed per websocket connection.
86+ defaultRotateEvery = 15 * time.Minute
87+ emptyUserListBackoff = 30 * time.Second
88+)
89+
7990 type JetstreamConsumer struct {
8091 client *jsc.Client
92+ cfg *jsc.ClientConfig
8193 logger *slog.Logger
8294 sched *jetstreamScheduler
8395 cursorStore *BatchCursorStore
96+ didProvider KnownDIDsProvider
8497 rewind time.Duration
98+ rotateEvery time.Duration
8599 }
86100
87-func NewJetstreamConsumer(jetstreamURL string, handler EventHandler, logger *slog.Logger, cursorStore CursorStore) *JetstreamConsumer {
101+func NewJetstreamConsumer(jetstreamURL string, handler EventHandler, logger *slog.Logger, cursorStore CursorStore, didProvider KnownDIDsProvider) *JetstreamConsumer {
88102 batchCursor := NewBatchCursorStore(cursorStore, 5*time.Second, logger)
89103
90104 sched := &jetstreamScheduler{
@@ -122,37 +136,72 @@ func NewJetstreamConsumer(jetstreamURL string, handler EventHandler, logger *slo
122136 return nil
123137 }
124138
125- rewind := 5 * time.Second
139+ rewind := defaultRewind
126140 if cursorStore == nil {
127141 rewind = 0
128142 }
129143
130144 return &JetstreamConsumer{
131145 client: c,
146+ cfg: config,
132147 logger: logger,
133148 sched: sched,
134149 cursorStore: batchCursor,
150+ didProvider: didProvider,
135151 rewind: rewind,
152+ rotateEvery: defaultRotateEvery,
136153 }
137154 }
138155
156+// refreshWantedDIDs narrows the subscription to the current set of known
157+// users. The jetstream client reads its config on every dial, so updates take
158+// effect on the next (re)connect; until then events of newly signed-up users
159+// are covered by PDS sync instead.
160+func (jc *JetstreamConsumer) refreshWantedDIDs(ctx context.Context) error {
161+ if jc.didProvider == nil {
162+ return nil
163+ }
164+ dids, err := jc.didProvider(ctx)
165+ if err != nil {
166+ return fmt.Errorf("listing known users: %w", err)
167+ }
168+ jc.cfg.WantedDids = dids
169+ jc.logger.Info("jetstream subscribed to known users", "count", len(dids))
170+ return nil
171+}
172+
139173 func (jc *JetstreamConsumer) Start(ctx context.Context) error {
140174 go jc.cursorStore.Run(ctx)
141175
142176 for {
143- var cursorPtr *int64
144- if jc.cursorStore != nil {
145- cur, err := jc.cursorStore.LoadCursor(ctx)
146- if err != nil {
147- jc.logger.Warn("failed to load cursor, starting from now", "error", err)
148- } else if cur != nil {
149- rewound := max(*cur-int64(jc.rewind/time.Microsecond), 0)
150- cursorPtr = &rewound
151- jc.logger.Info("resuming jetstream", "cursor_us", *cur, "rewound_us", rewound)
177+ if jc.didProvider != nil {
178+ if err := jc.refreshWantedDIDs(ctx); err != nil {
179+ jc.logger.Warn("keeping previous known-user filter", "error", err)
180+ } else if len(jc.cfg.WantedDids) == 0 {
181+ // Subscribing to nobody is pointless and would spin; wait for
182+ // users instead (backfill or sign-up creates them).
183+ jc.logger.Info("no known users yet; waiting before subscribing")
184+ select {
185+ case <-ctx.Done():
186+ return ctx.Err()
187+ case <-time.After(emptyUserListBackoff):
188+ }
189+ continue
152190 }
153191 }
154192
155- err := jc.client.ConnectAndRead(ctx, cursorPtr)
193+ var cursorPtr *int64
194+ if cur, err := jc.cursorStore.LoadCursor(ctx); err != nil {
195+ jc.logger.Warn("failed to load cursor, starting from now", "error", err)
196+ } else if cur != nil {
197+ rewound := max(*cur-int64(jc.rewind/time.Microsecond), 0)
198+ cursorPtr = &rewound
199+ jc.logger.Info("resuming jetstream", "cursor_us", *cur, "rewound_us", rewound)
200+ }
201+
202+ connCtx, cancel := context.WithTimeout(ctx, jc.rotateEvery)
203+ err := jc.client.ConnectAndRead(connCtx, cursorPtr)
204+ cancel()
156205 if ctx.Err() != nil {
157206 return ctx.Err()
158207 }
@@ -76,15 +76,29 @@ func (s *jetstreamScheduler) AddWork(ctx context.Context, _ string, evt *models.
76 76
77 func (s *jetstreamScheduler) Shutdown() {}77 func (s *jetstreamScheduler) Shutdown() {}
78 78
79+// KnownDIDsProvider returns the DIDs of all users Glean currently knows.
80+type KnownDIDsProvider func(ctx context.Context) ([]string, error)
81+
82+const (
83+ defaultRewind = 5 * time.Second
84+ // Connections are rotated regularly so an updated known-user filter takes
85+ // effect: wantedDids is fixed per websocket connection.
86+ defaultRotateEvery = 15 * time.Minute
87+ emptyUserListBackoff = 30 * time.Second
88+)
89+
79 type JetstreamConsumer struct {90 type JetstreamConsumer struct {
80 client *jsc.Client91 client *jsc.Client
92+ cfg *jsc.ClientConfig
81 logger *slog.Logger93 logger *slog.Logger
82 sched *jetstreamScheduler94 sched *jetstreamScheduler
83 cursorStore *BatchCursorStore95 cursorStore *BatchCursorStore
96+ didProvider KnownDIDsProvider
84 rewind time.Duration97 rewind time.Duration
98+ rotateEvery time.Duration
85 }99 }
86 100
87-func NewJetstreamConsumer(jetstreamURL string, handler EventHandler, logger *slog.Logger, cursorStore CursorStore) *JetstreamConsumer {101+func NewJetstreamConsumer(jetstreamURL string, handler EventHandler, logger *slog.Logger, cursorStore CursorStore, didProvider KnownDIDsProvider) *JetstreamConsumer {
88 batchCursor := NewBatchCursorStore(cursorStore, 5*time.Second, logger)102 batchCursor := NewBatchCursorStore(cursorStore, 5*time.Second, logger)
89 103
90 sched := &jetstreamScheduler{104 sched := &jetstreamScheduler{
@@ -122,37 +136,72 @@ func NewJetstreamConsumer(jetstreamURL string, handler EventHandler, logger *slo
122 return nil136 return nil
123 }137 }
124 138
125- rewind := 5 * time.Second139+ rewind := defaultRewind
126 if cursorStore == nil {140 if cursorStore == nil {
127 rewind = 0141 rewind = 0
128 }142 }
129 143
130 return &JetstreamConsumer{144 return &JetstreamConsumer{
131 client: c,145 client: c,
146+ cfg: config,
132 logger: logger,147 logger: logger,
133 sched: sched,148 sched: sched,
134 cursorStore: batchCursor,149 cursorStore: batchCursor,
150+ didProvider: didProvider,
135 rewind: rewind,151 rewind: rewind,
152+ rotateEvery: defaultRotateEvery,
136 }153 }
137 }154 }
138 155
156+// refreshWantedDIDs narrows the subscription to the current set of known
157+// users. The jetstream client reads its config on every dial, so updates take
158+// effect on the next (re)connect; until then events of newly signed-up users
159+// are covered by PDS sync instead.
160+func (jc *JetstreamConsumer) refreshWantedDIDs(ctx context.Context) error {
161+ if jc.didProvider == nil {
162+ return nil
163+ }
164+ dids, err := jc.didProvider(ctx)
165+ if err != nil {
166+ return fmt.Errorf("listing known users: %w", err)
167+ }
168+ jc.cfg.WantedDids = dids
169+ jc.logger.Info("jetstream subscribed to known users", "count", len(dids))
170+ return nil
171+}
172+
139 func (jc *JetstreamConsumer) Start(ctx context.Context) error {173 func (jc *JetstreamConsumer) Start(ctx context.Context) error {
140 go jc.cursorStore.Run(ctx)174 go jc.cursorStore.Run(ctx)
141 175
142 for {176 for {
143- var cursorPtr *int64177+ if jc.didProvider != nil {
144- if jc.cursorStore != nil {178+ if err := jc.refreshWantedDIDs(ctx); err != nil {
145- cur, err := jc.cursorStore.LoadCursor(ctx)179+ jc.logger.Warn("keeping previous known-user filter", "error", err)
146- if err != nil {180+ } else if len(jc.cfg.WantedDids) == 0 {
147- jc.logger.Warn("failed to load cursor, starting from now", "error", err)181+ // Subscribing to nobody is pointless and would spin; wait for
148- } else if cur != nil {182+ // users instead (backfill or sign-up creates them).
149- rewound := max(*cur-int64(jc.rewind/time.Microsecond), 0)183+ jc.logger.Info("no known users yet; waiting before subscribing")
150- cursorPtr = &rewound184+ select {
151- jc.logger.Info("resuming jetstream", "cursor_us", *cur, "rewound_us", rewound)185+ case <-ctx.Done():
186+ return ctx.Err()
187+ case <-time.After(emptyUserListBackoff):
188+ }
189+ continue
152 }190 }
153 }191 }
154 192
155- err := jc.client.ConnectAndRead(ctx, cursorPtr)193+ var cursorPtr *int64
194+ if cur, err := jc.cursorStore.LoadCursor(ctx); err != nil {
195+ jc.logger.Warn("failed to load cursor, starting from now", "error", err)
196+ } else if cur != nil {
197+ rewound := max(*cur-int64(jc.rewind/time.Microsecond), 0)
198+ cursorPtr = &rewound
199+ jc.logger.Info("resuming jetstream", "cursor_us", *cur, "rewound_us", rewound)
200+ }
201+
202+ connCtx, cancel := context.WithTimeout(ctx, jc.rotateEvery)
203+ err := jc.client.ConnectAndRead(connCtx, cursorPtr)
204+ cancel()
156 if ctx.Err() != nil {205 if ctx.Err() != nil {
157 return ctx.Err()206 return ctx.Err()
158 }207 }
added internal/atproto/jetstream_test.go +61 -0
new file mode 100644
@@ -0,0 +1,61 @@
1+package atproto
2+
3+import (
4+ "context"
5+ "errors"
6+ "io"
7+ "log/slog"
8+ "testing"
9+
10+ "gotest.tools/v3/assert"
11+)
12+
13+func testConsumer(t *testing.T, provider KnownDIDsProvider) *JetstreamConsumer {
14+ t.Helper()
15+ logger := slog.New(slog.NewTextHandler(io.Discard, nil))
16+ return NewJetstreamConsumer("wss://jetstream.example", func(context.Context, *Event) error { return nil }, logger, nil, provider)
17+}
18+
19+func TestRefreshWantedDIDs_UsesProvider(t *testing.T) {
20+ ctx := context.Background()
21+ jc := testConsumer(t, func(context.Context) ([]string, error) {
22+ return []string{"did:test:a", "did:test:b"}, nil
23+ })
24+
25+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
26+ assert.DeepEqual(t, []string{"did:test:a", "did:test:b"}, jc.cfg.WantedDids)
27+}
28+
29+func TestRefreshWantedDIDs_ErrorKeepsPreviousFilter(t *testing.T) {
30+ ctx := context.Background()
31+ first := true
32+ jc := testConsumer(t, func(context.Context) ([]string, error) {
33+ if first {
34+ first = false
35+ return []string{"did:test:a"}, nil
36+ }
37+ return nil, errors.New("db down")
38+ })
39+
40+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
41+ assert.Equal(t, 1, len(jc.cfg.WantedDids))
42+
43+ assert.ErrorContains(t, jc.refreshWantedDIDs(ctx), "db down")
44+ assert.Equal(t, 1, len(jc.cfg.WantedDids), "previous filter must survive provider failure")
45+}
46+
47+func TestRefreshWantedDIDs_EmptyListIsNotAnError(t *testing.T) {
48+ ctx := context.Background()
49+ jc := testConsumer(t, func(context.Context) ([]string, error) { return nil, nil })
50+
51+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
52+ assert.Equal(t, 0, len(jc.cfg.WantedDids))
53+}
54+
55+func TestRefreshWantedDIDs_NoProviderIsNoop(t *testing.T) {
56+ ctx := context.Background()
57+ jc := testConsumer(t, nil)
58+
59+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
60+ assert.Equal(t, 0, len(jc.cfg.WantedDids))
61+}
new file mode 100644
@@ -0,0 +1,61 @@
1+package atproto
2+
3+import (
4+ "context"
5+ "errors"
6+ "io"
7+ "log/slog"
8+ "testing"
9+
10+ "gotest.tools/v3/assert"
11+)
12+
13+func testConsumer(t *testing.T, provider KnownDIDsProvider) *JetstreamConsumer {
14+ t.Helper()
15+ logger := slog.New(slog.NewTextHandler(io.Discard, nil))
16+ return NewJetstreamConsumer("wss://jetstream.example", func(context.Context, *Event) error { return nil }, logger, nil, provider)
17+}
18+
19+func TestRefreshWantedDIDs_UsesProvider(t *testing.T) {
20+ ctx := context.Background()
21+ jc := testConsumer(t, func(context.Context) ([]string, error) {
22+ return []string{"did:test:a", "did:test:b"}, nil
23+ })
24+
25+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
26+ assert.DeepEqual(t, []string{"did:test:a", "did:test:b"}, jc.cfg.WantedDids)
27+}
28+
29+func TestRefreshWantedDIDs_ErrorKeepsPreviousFilter(t *testing.T) {
30+ ctx := context.Background()
31+ first := true
32+ jc := testConsumer(t, func(context.Context) ([]string, error) {
33+ if first {
34+ first = false
35+ return []string{"did:test:a"}, nil
36+ }
37+ return nil, errors.New("db down")
38+ })
39+
40+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
41+ assert.Equal(t, 1, len(jc.cfg.WantedDids))
42+
43+ assert.ErrorContains(t, jc.refreshWantedDIDs(ctx), "db down")
44+ assert.Equal(t, 1, len(jc.cfg.WantedDids), "previous filter must survive provider failure")
45+}
46+
47+func TestRefreshWantedDIDs_EmptyListIsNotAnError(t *testing.T) {
48+ ctx := context.Background()
49+ jc := testConsumer(t, func(context.Context) ([]string, error) { return nil, nil })
50+
51+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
52+ assert.Equal(t, 0, len(jc.cfg.WantedDids))
53+}
54+
55+func TestRefreshWantedDIDs_NoProviderIsNoop(t *testing.T) {
56+ ctx := context.Background()
57+ jc := testConsumer(t, nil)
58+
59+ assert.NilError(t, jc.refreshWantedDIDs(ctx))
60+ assert.Equal(t, 0, len(jc.cfg.WantedDids))
61+}
modified internal/cluster/cron.go +13 -10
@@ -13,19 +13,22 @@ import (
1313 // follow distances, signal profiles, auto-dismiss, recommendation precomputation)
1414 // on a fixed interval.
1515 type Cron struct {
16- engine *Engine
17- interval time.Duration
18- logger *slog.Logger
19- dbs *db.Store
16+ engine *Engine
17+ interval time.Duration
18+ articleRetentionDays int
19+ logger *slog.Logger
20+ dbs *db.Store
2021 }
2122
2223 // NewCron creates a new cron runner with the given engine and interval.
23-func NewCron(engine *Engine, interval time.Duration, logger *slog.Logger, dbs *db.Store) *Cron {
24+// articleRetentionDays bounds how long articles are kept.
25+func NewCron(engine *Engine, interval time.Duration, logger *slog.Logger, dbs *db.Store, articleRetentionDays int) *Cron {
2426 return &Cron{
25- engine: engine,
26- interval: interval,
27- logger: logger,
28- dbs: dbs,
27+ engine: engine,
28+ interval: interval,
29+ articleRetentionDays: articleRetentionDays,
30+ logger: logger,
31+ dbs: dbs,
2932 }
3033 }
3134
@@ -71,7 +74,7 @@ func (c *Cron) Run(ctx context.Context) error {
7174 c.engine.mu.Unlock()
7275 }
7376
74- if err := c.dbs.RunMaintenance(ctx, 90); err != nil {
77+ if err := c.dbs.RunMaintenance(ctx, 90, c.articleRetentionDays); err != nil {
7578 c.logger.Error("db maintenance failed", "error", err)
7679 }
7780
@@ -13,19 +13,22 @@ import (
13 // follow distances, signal profiles, auto-dismiss, recommendation precomputation)13 // follow distances, signal profiles, auto-dismiss, recommendation precomputation)
14 // on a fixed interval.14 // on a fixed interval.
15 type Cron struct {15 type Cron struct {
16- engine *Engine16+ engine *Engine
17- interval time.Duration17+ interval time.Duration
18- logger *slog.Logger18+ articleRetentionDays int
19- dbs *db.Store19+ logger *slog.Logger
20+ dbs *db.Store
20 }21 }
21 22
22 // NewCron creates a new cron runner with the given engine and interval.23 // NewCron creates a new cron runner with the given engine and interval.
23-func NewCron(engine *Engine, interval time.Duration, logger *slog.Logger, dbs *db.Store) *Cron {24+// articleRetentionDays bounds how long articles are kept.
25+func NewCron(engine *Engine, interval time.Duration, logger *slog.Logger, dbs *db.Store, articleRetentionDays int) *Cron {
24 return &Cron{26 return &Cron{
25- engine: engine,27+ engine: engine,
26- interval: interval,28+ interval: interval,
27- logger: logger,29+ articleRetentionDays: articleRetentionDays,
28- dbs: dbs,30+ logger: logger,
31+ dbs: dbs,
29 }32 }
30 }33 }
31 34
@@ -71,7 +74,7 @@ func (c *Cron) Run(ctx context.Context) error {
71 c.engine.mu.Unlock()74 c.engine.mu.Unlock()
72 }75 }
73 76
74- if err := c.dbs.RunMaintenance(ctx, 90); err != nil {77+ if err := c.dbs.RunMaintenance(ctx, 90, c.articleRetentionDays); err != nil {
75 c.logger.Error("db maintenance failed", "error", err)78 c.logger.Error("db maintenance failed", "error", err)
76 }79 }
77 80
modified internal/db/db.go +8 -1
@@ -126,12 +126,19 @@ func Open(basePath string) (*Store, error) {
126126 }, nil
127127 }
128128
129-func (s *Store) RunMaintenance(ctx context.Context, impressionMaxAgeDays int) error {
129+// RunMaintenance prunes data that is no longer needed: recommendation
130+// impressions older than impressionMaxAgeDays, articles older than
131+// articleRetentionDays, and freed pages (incremental vacuum).
132+func (s *Store) RunMaintenance(ctx context.Context, impressionMaxAgeDays, articleRetentionDays int) error {
130133 cutoff := time.Now().AddDate(0, 0, -impressionMaxAgeDays).Format(time.RFC3339)
131134 if _, err := s.db.ExecContext(ctx, `DELETE FROM main.recommendation_impressions WHERE first_shown_at < ?`, cutoff); err != nil {
132135 return fmt.Errorf("prune impressions: %w", err)
133136 }
134137
138+ if _, err := s.PurgeExpiredArticles(ctx, articleRetentionDays); err != nil {
139+ return fmt.Errorf("purge expired articles: %w", err)
140+ }
141+
135142 for _, schema := range []string{"main", "articles", "recs"} {
136143 if _, err := s.db.ExecContext(ctx, fmt.Sprintf("PRAGMA %s.incremental_vacuum", schema)); err != nil {
137144 return fmt.Errorf("incremental_vacuum %s: %w", schema, err)
@@ -126,12 +126,19 @@ func Open(basePath string) (*Store, error) {
126 }, nil126 }, nil
127 }127 }
128 128
129-func (s *Store) RunMaintenance(ctx context.Context, impressionMaxAgeDays int) error {129+// RunMaintenance prunes data that is no longer needed: recommendation
130+// impressions older than impressionMaxAgeDays, articles older than
131+// articleRetentionDays, and freed pages (incremental vacuum).
132+func (s *Store) RunMaintenance(ctx context.Context, impressionMaxAgeDays, articleRetentionDays int) error {
130 cutoff := time.Now().AddDate(0, 0, -impressionMaxAgeDays).Format(time.RFC3339)133 cutoff := time.Now().AddDate(0, 0, -impressionMaxAgeDays).Format(time.RFC3339)
131 if _, err := s.db.ExecContext(ctx, `DELETE FROM main.recommendation_impressions WHERE first_shown_at < ?`, cutoff); err != nil {134 if _, err := s.db.ExecContext(ctx, `DELETE FROM main.recommendation_impressions WHERE first_shown_at < ?`, cutoff); err != nil {
132 return fmt.Errorf("prune impressions: %w", err)135 return fmt.Errorf("prune impressions: %w", err)
133 }136 }
134 137
138+ if _, err := s.PurgeExpiredArticles(ctx, articleRetentionDays); err != nil {
139+ return fmt.Errorf("purge expired articles: %w", err)
140+ }
141+
135 for _, schema := range []string{"main", "articles", "recs"} {142 for _, schema := range []string{"main", "articles", "recs"} {
136 if _, err := s.db.ExecContext(ctx, fmt.Sprintf("PRAGMA %s.incremental_vacuum", schema)); err != nil {143 if _, err := s.db.ExecContext(ctx, fmt.Sprintf("PRAGMA %s.incremental_vacuum", schema)); err != nil {
137 return fmt.Errorf("incremental_vacuum %s: %w", schema, err)144 return fmt.Errorf("incremental_vacuum %s: %w", schema, err)
added internal/db/retention.go +144 -0
new file mode 100644
@@ -0,0 +1,144 @@
1+package db
2+
3+import (
4+ "context"
5+ "fmt"
6+ "time"
7+)
8+
9+// Retention keeps the database from growing without bound. Glean used to ingest
10+// network-wide ATProto activity (jetstream) and keep full article bodies
11+// forever; once SQLite cannot write, everything that persists state breaks —
12+// including OAuth sign-in.
13+
14+const purgeBatchSize = 5000
15+
16+// PurgeStats reports the rows removed per table by a purge pass. Tables with
17+// nothing to purge are omitted.
18+type PurgeStats map[string]int64
19+
20+// Total returns the number of rows removed across all tables.
21+func (p PurgeStats) Total() int64 {
22+ var total int64
23+ for _, n := range p {
24+ total += n
25+ }
26+ return total
27+}
28+
29+// PurgeUnknownUserRows deletes likes, follows, subscriptions, annotations,
30+// read state, impressions and dismissals whose owner DID is not in the users
31+// table. Jetstream used to deliver events for the whole network; only rows
32+// owned by known users are meaningful for feeds and recommendations.
33+//
34+// Deletes run in batches so the WAL cannot grow by the full table size on a
35+// nearly-full volume.
36+func (s *Store) PurgeUnknownUserRows(ctx context.Context) (PurgeStats, error) {
37+ stats := PurgeStats{}
38+ // Follows first: that table is the one that filled the users DB.
39+ queries := []struct {
40+ name string
41+ query string
42+ }{
43+ {"follows", `DELETE FROM follows WHERE rowid IN (SELECT rowid FROM follows WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
44+ {"likes", `DELETE FROM articles.likes WHERE rowid IN (SELECT rowid FROM articles.likes WHERE author_did NOT IN (SELECT did FROM users) LIMIT ?)`},
45+ {"annotations", `DELETE FROM articles.annotations WHERE rowid IN (SELECT rowid FROM articles.annotations WHERE author_did NOT IN (SELECT did FROM users) LIMIT ?)`},
46+ {"subscriptions", `DELETE FROM articles.subscriptions WHERE rowid IN (SELECT rowid FROM articles.subscriptions WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
47+ {"read_state", `DELETE FROM articles.read_state WHERE rowid IN (SELECT rowid FROM articles.read_state WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
48+ {"recommendation_impressions", `DELETE FROM recommendation_impressions WHERE rowid IN (SELECT rowid FROM recommendation_impressions WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
49+ {"dismissed_recommendations", `DELETE FROM dismissed_recommendations WHERE rowid IN (SELECT rowid FROM dismissed_recommendations WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
50+ }
51+ for _, q := range queries {
52+ n, err := s.deleteInBatches(ctx, q.query)
53+ if err != nil {
54+ return stats, fmt.Errorf("purge %s: %w", q.name, err)
55+ }
56+ if n > 0 {
57+ stats[q.name] = n
58+ }
59+ }
60+ return stats, nil
61+}
62+
63+func (s *Store) deleteInBatches(ctx context.Context, query string) (int64, error) {
64+ var total int64
65+ for {
66+ if err := ctx.Err(); err != nil {
67+ return total, err
68+ }
69+ res, err := s.db.ExecContext(ctx, query, purgeBatchSize)
70+ if err != nil {
71+ return total, err
72+ }
73+ n, err := res.RowsAffected()
74+ if err != nil {
75+ return total, err
76+ }
77+ total += n
78+ if n < int64(purgeBatchSize) {
79+ return total, nil
80+ }
81+ }
82+}
83+
84+// PurgeExpiredArticles deletes articles older than maxAgeDays — by published
85+// date, falling back to fetched_at for undated items — together with their
86+// read-state rows. It returns the number of articles removed.
87+func (s *Store) PurgeExpiredArticles(ctx context.Context, maxAgeDays int) (int64, error) {
88+ if maxAgeDays <= 0 {
89+ return 0, nil
90+ }
91+ cutoff := time.Now().AddDate(0, 0, -maxAgeDays)
92+ const stale = `(published IS NOT NULL AND published < ?1) OR (published IS NULL AND fetched_at < ?1)`
93+
94+ if _, err := s.db.ExecContext(ctx,
95+ `DELETE FROM articles.read_state WHERE article_id IN (SELECT id FROM articles.articles WHERE `+stale+`)`,
96+ cutoff,
97+ ); err != nil {
98+ return 0, fmt.Errorf("purge read state of old articles: %w", err)
99+ }
100+
101+ var total int64
102+ for {
103+ if err := ctx.Err(); err != nil {
104+ return total, err
105+ }
106+ res, err := s.db.ExecContext(ctx,
107+ `DELETE FROM articles.articles WHERE id IN (SELECT id FROM articles.articles WHERE `+stale+` LIMIT ?2)`,
108+ cutoff, purgeBatchSize,
109+ )
110+ if err != nil {
111+ return total, fmt.Errorf("purge old articles: %w", err)
112+ }
113+ n, _ := res.RowsAffected()
114+ total += n
115+ if n < int64(purgeBatchSize) {
116+ return total, nil
117+ }
118+ }
119+}
120+
121+// ReclaimSpace shrinks the database files after large purges. VACUUM is
122+// pointed at the container's ephemeral /tmp so it can complete even when the
123+// data volume itself is full; incremental vacuum is a fallback if VACUUM
124+// cannot run.
125+func (s *Store) ReclaimSpace(ctx context.Context) error {
126+ if _, err := s.db.ExecContext(ctx, `PRAGMA temp_store_directory = '/tmp'`); err != nil {
127+ return fmt.Errorf("temp_store_directory: %w", err)
128+ }
129+ var first error
130+ for _, schema := range []string{"main", "articles", "recs"} {
131+ if _, err := s.db.ExecContext(ctx, fmt.Sprintf(`PRAGMA %s.wal_checkpoint(TRUNCATE)`, schema)); err != nil && first == nil {
132+ first = fmt.Errorf("wal_checkpoint %s: %w", schema, err)
133+ }
134+ if _, err := s.db.ExecContext(ctx, fmt.Sprintf(`VACUUM %s`, schema)); err != nil {
135+ if first == nil {
136+ first = fmt.Errorf("vacuum %s: %w", schema, err)
137+ }
138+ if _, err := s.db.ExecContext(ctx, fmt.Sprintf(`PRAGMA %s.incremental_vacuum`, schema)); err != nil && first == nil {
139+ first = fmt.Errorf("incremental_vacuum %s: %w", schema, err)
140+ }
141+ }
142+ }
143+ return first
144+}
new file mode 100644
@@ -0,0 +1,144 @@
1+package db
2+
3+import (
4+ "context"
5+ "fmt"
6+ "time"
7+)
8+
9+// Retention keeps the database from growing without bound. Glean used to ingest
10+// network-wide ATProto activity (jetstream) and keep full article bodies
11+// forever; once SQLite cannot write, everything that persists state breaks —
12+// including OAuth sign-in.
13+
14+const purgeBatchSize = 5000
15+
16+// PurgeStats reports the rows removed per table by a purge pass. Tables with
17+// nothing to purge are omitted.
18+type PurgeStats map[string]int64
19+
20+// Total returns the number of rows removed across all tables.
21+func (p PurgeStats) Total() int64 {
22+ var total int64
23+ for _, n := range p {
24+ total += n
25+ }
26+ return total
27+}
28+
29+// PurgeUnknownUserRows deletes likes, follows, subscriptions, annotations,
30+// read state, impressions and dismissals whose owner DID is not in the users
31+// table. Jetstream used to deliver events for the whole network; only rows
32+// owned by known users are meaningful for feeds and recommendations.
33+//
34+// Deletes run in batches so the WAL cannot grow by the full table size on a
35+// nearly-full volume.
36+func (s *Store) PurgeUnknownUserRows(ctx context.Context) (PurgeStats, error) {
37+ stats := PurgeStats{}
38+ // Follows first: that table is the one that filled the users DB.
39+ queries := []struct {
40+ name string
41+ query string
42+ }{
43+ {"follows", `DELETE FROM follows WHERE rowid IN (SELECT rowid FROM follows WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
44+ {"likes", `DELETE FROM articles.likes WHERE rowid IN (SELECT rowid FROM articles.likes WHERE author_did NOT IN (SELECT did FROM users) LIMIT ?)`},
45+ {"annotations", `DELETE FROM articles.annotations WHERE rowid IN (SELECT rowid FROM articles.annotations WHERE author_did NOT IN (SELECT did FROM users) LIMIT ?)`},
46+ {"subscriptions", `DELETE FROM articles.subscriptions WHERE rowid IN (SELECT rowid FROM articles.subscriptions WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
47+ {"read_state", `DELETE FROM articles.read_state WHERE rowid IN (SELECT rowid FROM articles.read_state WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
48+ {"recommendation_impressions", `DELETE FROM recommendation_impressions WHERE rowid IN (SELECT rowid FROM recommendation_impressions WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
49+ {"dismissed_recommendations", `DELETE FROM dismissed_recommendations WHERE rowid IN (SELECT rowid FROM dismissed_recommendations WHERE user_did NOT IN (SELECT did FROM users) LIMIT ?)`},
50+ }
51+ for _, q := range queries {
52+ n, err := s.deleteInBatches(ctx, q.query)
53+ if err != nil {
54+ return stats, fmt.Errorf("purge %s: %w", q.name, err)
55+ }
56+ if n > 0 {
57+ stats[q.name] = n
58+ }
59+ }
60+ return stats, nil
61+}
62+
63+func (s *Store) deleteInBatches(ctx context.Context, query string) (int64, error) {
64+ var total int64
65+ for {
66+ if err := ctx.Err(); err != nil {
67+ return total, err
68+ }
69+ res, err := s.db.ExecContext(ctx, query, purgeBatchSize)
70+ if err != nil {
71+ return total, err
72+ }
73+ n, err := res.RowsAffected()
74+ if err != nil {
75+ return total, err
76+ }
77+ total += n
78+ if n < int64(purgeBatchSize) {
79+ return total, nil
80+ }
81+ }
82+}
83+
84+// PurgeExpiredArticles deletes articles older than maxAgeDays — by published
85+// date, falling back to fetched_at for undated items — together with their
86+// read-state rows. It returns the number of articles removed.
87+func (s *Store) PurgeExpiredArticles(ctx context.Context, maxAgeDays int) (int64, error) {
88+ if maxAgeDays <= 0 {
89+ return 0, nil
90+ }
91+ cutoff := time.Now().AddDate(0, 0, -maxAgeDays)
92+ const stale = `(published IS NOT NULL AND published < ?1) OR (published IS NULL AND fetched_at < ?1)`
93+
94+ if _, err := s.db.ExecContext(ctx,
95+ `DELETE FROM articles.read_state WHERE article_id IN (SELECT id FROM articles.articles WHERE `+stale+`)`,
96+ cutoff,
97+ ); err != nil {
98+ return 0, fmt.Errorf("purge read state of old articles: %w", err)
99+ }
100+
101+ var total int64
102+ for {
103+ if err := ctx.Err(); err != nil {
104+ return total, err
105+ }
106+ res, err := s.db.ExecContext(ctx,
107+ `DELETE FROM articles.articles WHERE id IN (SELECT id FROM articles.articles WHERE `+stale+` LIMIT ?2)`,
108+ cutoff, purgeBatchSize,
109+ )
110+ if err != nil {
111+ return total, fmt.Errorf("purge old articles: %w", err)
112+ }
113+ n, _ := res.RowsAffected()
114+ total += n
115+ if n < int64(purgeBatchSize) {
116+ return total, nil
117+ }
118+ }
119+}
120+
121+// ReclaimSpace shrinks the database files after large purges. VACUUM is
122+// pointed at the container's ephemeral /tmp so it can complete even when the
123+// data volume itself is full; incremental vacuum is a fallback if VACUUM
124+// cannot run.
125+func (s *Store) ReclaimSpace(ctx context.Context) error {
126+ if _, err := s.db.ExecContext(ctx, `PRAGMA temp_store_directory = '/tmp'`); err != nil {
127+ return fmt.Errorf("temp_store_directory: %w", err)
128+ }
129+ var first error
130+ for _, schema := range []string{"main", "articles", "recs"} {
131+ if _, err := s.db.ExecContext(ctx, fmt.Sprintf(`PRAGMA %s.wal_checkpoint(TRUNCATE)`, schema)); err != nil && first == nil {
132+ first = fmt.Errorf("wal_checkpoint %s: %w", schema, err)
133+ }
134+ if _, err := s.db.ExecContext(ctx, fmt.Sprintf(`VACUUM %s`, schema)); err != nil {
135+ if first == nil {
136+ first = fmt.Errorf("vacuum %s: %w", schema, err)
137+ }
138+ if _, err := s.db.ExecContext(ctx, fmt.Sprintf(`PRAGMA %s.incremental_vacuum`, schema)); err != nil && first == nil {
139+ first = fmt.Errorf("incremental_vacuum %s: %w", schema, err)
140+ }
141+ }
142+ }
143+ return first
144+}
added internal/db/retention_test.go +133 -0
new file mode 100644
@@ -0,0 +1,133 @@
1+package db
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "gotest.tools/v3/assert"
9+)
10+
11+func seedRetentionFixtures(t *testing.T, ctx context.Context, dbs *Store) {
12+ t.Helper()
13+ sql := dbs.SQLDB()
14+
15+ // Known user + an unknown one whose rows must not survive purges.
16+ _, err := sql.ExecContext(ctx, `INSERT INTO users (did) VALUES ('did:test:known')`)
17+ assert.NilError(t, err)
18+
19+ insertArticle := func(t *testing.T, id int64, feedURL, guid string, published time.Time) {
20+ t.Helper()
21+ _, err := sql.ExecContext(ctx, `
22+ INSERT INTO articles.articles (id, feed_url, guid, title, published)
23+ VALUES (?, ?, ?, 't', ?)`,
24+ id, feedURL, guid, published)
25+ assert.NilError(t, err)
26+ }
27+
28+ now := time.Now()
29+ insertArticle(t, 1, "https://f.example/rss", "old-1", now.AddDate(0, 0, -60))
30+ insertArticle(t, 2, "https://f.example/rss", "fresh-2", now.AddDate(0, 0, -5))
31+
32+ for _, tc := range []struct {
33+ did string
34+ articleID int64
35+ }{
36+ {"did:test:known", 2},
37+ {"did:test:unknown", 1},
38+ } {
39+ _, err = sql.ExecContext(ctx,
40+ `INSERT INTO articles.read_state (user_did, article_id, is_read) VALUES (?, ?, 1)`,
41+ tc.did, tc.articleID)
42+ assert.NilError(t, err)
43+ }
44+
45+ fixtures := []struct {
46+ query string
47+ args []any
48+ }{
49+ {`INSERT INTO articles.likes (uri, author_did, feed_url, article_url, created_at) VALUES (?, ?, 'f', 'a', CURRENT_TIMESTAMP)`, []any{"at://did:test:unknown/at.glean.like/1", "did:test:unknown"}},
50+ {`INSERT INTO articles.annotations (uri, author_did, feed_url, article_url, created_at) VALUES (?, ?, 'f', 'a', CURRENT_TIMESTAMP)`, []any{"at://did:test:unknown/at.glean.annotation/1", "did:test:unknown"}},
51+ {`INSERT INTO articles.subscriptions (user_did, feed_url) VALUES (?, 'f')`, []any{"did:test:unknown"}},
52+ {`INSERT INTO follows (user_did, target_did) VALUES (?, 'did:test:known')`, []any{"did:test:unknown"}},
53+ {`INSERT INTO recommendation_impressions (user_did, target_type, target_id) VALUES (?, 'feed', 'f')`, []any{"did:test:unknown"}},
54+ {`INSERT INTO dismissed_recommendations (user_did, target_type, target_id) VALUES (?, 'feed', 'f')`, []any{"did:test:unknown"}},
55+ {`INSERT INTO articles.likes (uri, author_did, feed_url, article_url, created_at) VALUES (?, 'did:test:known', 'f', 'a', CURRENT_TIMESTAMP)`, []any{"at://did:test:known/at.glean.like/1"}},
56+ }
57+ for _, f := range fixtures {
58+ _, err = sql.ExecContext(ctx, f.query, f.args...)
59+ assert.NilError(t, err)
60+ }
61+}
62+
63+func count(t *testing.T, ctx context.Context, dbs *Store, query string, args ...any) int64 {
64+ t.Helper()
65+ var n int64
66+ err := dbs.SQLDB().QueryRowContext(ctx, query, args...).Scan(&n)
67+ assert.NilError(t, err)
68+ return n
69+}
70+
71+func TestPurgeUnknownUserRows_RemovesOnlyUnknownUserRows(t *testing.T) {
72+ ctx := context.Background()
73+ dbs := setupTestDB(t)
74+ seedRetentionFixtures(t, ctx, dbs)
75+
76+ stats, err := dbs.PurgeUnknownUserRows(ctx)
77+ assert.NilError(t, err)
78+ assert.Equal(t, int64(7), stats.Total())
79+
80+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.likes WHERE author_did = 'did:test:unknown'`))
81+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.likes WHERE author_did = 'did:test:known'`))
82+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.annotations WHERE author_did = 'did:test:unknown'`))
83+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.subscriptions WHERE user_did = 'did:test:unknown'`))
84+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM follows WHERE user_did = 'did:test:unknown'`))
85+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM recommendation_impressions WHERE user_did = 'did:test:unknown'`))
86+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM dismissed_recommendations WHERE user_did = 'did:test:unknown'`))
87+ // Read state of known users survives.
88+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.read_state WHERE user_did = 'did:test:known'`))
89+}
90+
91+func TestPurgeExpiredArticles_DeletesOldArticlesAndTheirReadState(t *testing.T) {
92+ ctx := context.Background()
93+ dbs := setupTestDB(t)
94+ seedRetentionFixtures(t, ctx, dbs)
95+
96+ n, err := dbs.PurgeExpiredArticles(ctx, 30)
97+ assert.NilError(t, err)
98+ assert.Equal(t, int64(1), n)
99+
100+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'old-1'`))
101+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'fresh-2'`))
102+ // The deleted article's read state goes with it; the recent one stays.
103+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.read_state r JOIN articles.articles a ON a.id = r.article_id WHERE a.guid = 'old-1'`))
104+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.read_state WHERE article_id = 2`))
105+}
106+
107+func TestPurgeExpiredArticles_ZeroDaysDisables(t *testing.T) {
108+ ctx := context.Background()
109+ dbs := setupTestDB(t)
110+ seedRetentionFixtures(t, ctx, dbs)
111+
112+ n, err := dbs.PurgeExpiredArticles(ctx, 0)
113+ assert.NilError(t, err)
114+ assert.Equal(t, int64(0), n)
115+ assert.Equal(t, int64(2), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles`))
116+}
117+
118+func TestReclaimSpace_RunsClean(t *testing.T) {
119+ ctx := context.Background()
120+ dbs := setupTestDB(t)
121+ seedRetentionFixtures(t, ctx, dbs)
122+
123+ assert.NilError(t, dbs.ReclaimSpace(ctx))
124+}
125+
126+func TestRunMaintenance_AppliesArticleRetention(t *testing.T) {
127+ ctx := context.Background()
128+ dbs := setupTestDB(t)
129+ seedRetentionFixtures(t, ctx, dbs)
130+
131+ assert.NilError(t, dbs.RunMaintenance(ctx, 90, 30))
132+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles`))
133+}
new file mode 100644
@@ -0,0 +1,133 @@
1+package db
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "gotest.tools/v3/assert"
9+)
10+
11+func seedRetentionFixtures(t *testing.T, ctx context.Context, dbs *Store) {
12+ t.Helper()
13+ sql := dbs.SQLDB()
14+
15+ // Known user + an unknown one whose rows must not survive purges.
16+ _, err := sql.ExecContext(ctx, `INSERT INTO users (did) VALUES ('did:test:known')`)
17+ assert.NilError(t, err)
18+
19+ insertArticle := func(t *testing.T, id int64, feedURL, guid string, published time.Time) {
20+ t.Helper()
21+ _, err := sql.ExecContext(ctx, `
22+ INSERT INTO articles.articles (id, feed_url, guid, title, published)
23+ VALUES (?, ?, ?, 't', ?)`,
24+ id, feedURL, guid, published)
25+ assert.NilError(t, err)
26+ }
27+
28+ now := time.Now()
29+ insertArticle(t, 1, "https://f.example/rss", "old-1", now.AddDate(0, 0, -60))
30+ insertArticle(t, 2, "https://f.example/rss", "fresh-2", now.AddDate(0, 0, -5))
31+
32+ for _, tc := range []struct {
33+ did string
34+ articleID int64
35+ }{
36+ {"did:test:known", 2},
37+ {"did:test:unknown", 1},
38+ } {
39+ _, err = sql.ExecContext(ctx,
40+ `INSERT INTO articles.read_state (user_did, article_id, is_read) VALUES (?, ?, 1)`,
41+ tc.did, tc.articleID)
42+ assert.NilError(t, err)
43+ }
44+
45+ fixtures := []struct {
46+ query string
47+ args []any
48+ }{
49+ {`INSERT INTO articles.likes (uri, author_did, feed_url, article_url, created_at) VALUES (?, ?, 'f', 'a', CURRENT_TIMESTAMP)`, []any{"at://did:test:unknown/at.glean.like/1", "did:test:unknown"}},
50+ {`INSERT INTO articles.annotations (uri, author_did, feed_url, article_url, created_at) VALUES (?, ?, 'f', 'a', CURRENT_TIMESTAMP)`, []any{"at://did:test:unknown/at.glean.annotation/1", "did:test:unknown"}},
51+ {`INSERT INTO articles.subscriptions (user_did, feed_url) VALUES (?, 'f')`, []any{"did:test:unknown"}},
52+ {`INSERT INTO follows (user_did, target_did) VALUES (?, 'did:test:known')`, []any{"did:test:unknown"}},
53+ {`INSERT INTO recommendation_impressions (user_did, target_type, target_id) VALUES (?, 'feed', 'f')`, []any{"did:test:unknown"}},
54+ {`INSERT INTO dismissed_recommendations (user_did, target_type, target_id) VALUES (?, 'feed', 'f')`, []any{"did:test:unknown"}},
55+ {`INSERT INTO articles.likes (uri, author_did, feed_url, article_url, created_at) VALUES (?, 'did:test:known', 'f', 'a', CURRENT_TIMESTAMP)`, []any{"at://did:test:known/at.glean.like/1"}},
56+ }
57+ for _, f := range fixtures {
58+ _, err = sql.ExecContext(ctx, f.query, f.args...)
59+ assert.NilError(t, err)
60+ }
61+}
62+
63+func count(t *testing.T, ctx context.Context, dbs *Store, query string, args ...any) int64 {
64+ t.Helper()
65+ var n int64
66+ err := dbs.SQLDB().QueryRowContext(ctx, query, args...).Scan(&n)
67+ assert.NilError(t, err)
68+ return n
69+}
70+
71+func TestPurgeUnknownUserRows_RemovesOnlyUnknownUserRows(t *testing.T) {
72+ ctx := context.Background()
73+ dbs := setupTestDB(t)
74+ seedRetentionFixtures(t, ctx, dbs)
75+
76+ stats, err := dbs.PurgeUnknownUserRows(ctx)
77+ assert.NilError(t, err)
78+ assert.Equal(t, int64(7), stats.Total())
79+
80+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.likes WHERE author_did = 'did:test:unknown'`))
81+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.likes WHERE author_did = 'did:test:known'`))
82+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.annotations WHERE author_did = 'did:test:unknown'`))
83+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.subscriptions WHERE user_did = 'did:test:unknown'`))
84+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM follows WHERE user_did = 'did:test:unknown'`))
85+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM recommendation_impressions WHERE user_did = 'did:test:unknown'`))
86+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM dismissed_recommendations WHERE user_did = 'did:test:unknown'`))
87+ // Read state of known users survives.
88+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.read_state WHERE user_did = 'did:test:known'`))
89+}
90+
91+func TestPurgeExpiredArticles_DeletesOldArticlesAndTheirReadState(t *testing.T) {
92+ ctx := context.Background()
93+ dbs := setupTestDB(t)
94+ seedRetentionFixtures(t, ctx, dbs)
95+
96+ n, err := dbs.PurgeExpiredArticles(ctx, 30)
97+ assert.NilError(t, err)
98+ assert.Equal(t, int64(1), n)
99+
100+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'old-1'`))
101+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'fresh-2'`))
102+ // The deleted article's read state goes with it; the recent one stays.
103+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.read_state r JOIN articles.articles a ON a.id = r.article_id WHERE a.guid = 'old-1'`))
104+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.read_state WHERE article_id = 2`))
105+}
106+
107+func TestPurgeExpiredArticles_ZeroDaysDisables(t *testing.T) {
108+ ctx := context.Background()
109+ dbs := setupTestDB(t)
110+ seedRetentionFixtures(t, ctx, dbs)
111+
112+ n, err := dbs.PurgeExpiredArticles(ctx, 0)
113+ assert.NilError(t, err)
114+ assert.Equal(t, int64(0), n)
115+ assert.Equal(t, int64(2), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles`))
116+}
117+
118+func TestReclaimSpace_RunsClean(t *testing.T) {
119+ ctx := context.Background()
120+ dbs := setupTestDB(t)
121+ seedRetentionFixtures(t, ctx, dbs)
122+
123+ assert.NilError(t, dbs.ReclaimSpace(ctx))
124+}
125+
126+func TestRunMaintenance_AppliesArticleRetention(t *testing.T) {
127+ ctx := context.Background()
128+ dbs := setupTestDB(t)
129+ seedRetentionFixtures(t, ctx, dbs)
130+
131+ assert.NilError(t, dbs.RunMaintenance(ctx, 90, 30))
132+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles`))
133+}
modified internal/db/user.go +19 -0
@@ -87,6 +87,25 @@ func (s *UserStore) UserDIDs(ctx context.Context) (map[string]bool, error) {
8787 return dids, rows.Err()
8888 }
8989
90+// UserDIDList returns the DIDs of all known users.
91+func (s *UserStore) UserDIDList(ctx context.Context) ([]string, error) {
92+ rows, err := s.db.QueryContext(ctx, `SELECT did FROM users`)
93+ if err != nil {
94+ return nil, err
95+ }
96+ defer rows.Close()
97+
98+ var dids []string
99+ for rows.Next() {
100+ var did string
101+ if err := rows.Scan(&did); err != nil {
102+ return nil, err
103+ }
104+ dids = append(dids, did)
105+ }
106+ return dids, rows.Err()
107+}
108+
90109 func (s *UserStore) ListUsers(ctx context.Context) ([]*User, error) {
91110 rows, err := s.db.QueryContext(ctx, `
92111 SELECT did, indexed_at, updated_at, follows_dirty
@@ -87,6 +87,25 @@ func (s *UserStore) UserDIDs(ctx context.Context) (map[string]bool, error) {
87 return dids, rows.Err()87 return dids, rows.Err()
88 }88 }
89 89
90+// UserDIDList returns the DIDs of all known users.
91+func (s *UserStore) UserDIDList(ctx context.Context) ([]string, error) {
92+ rows, err := s.db.QueryContext(ctx, `SELECT did FROM users`)
93+ if err != nil {
94+ return nil, err
95+ }
96+ defer rows.Close()
97+
98+ var dids []string
99+ for rows.Next() {
100+ var did string
101+ if err := rows.Scan(&did); err != nil {
102+ return nil, err
103+ }
104+ dids = append(dids, did)
105+ }
106+ return dids, rows.Err()
107+}
108+
90 func (s *UserStore) ListUsers(ctx context.Context) ([]*User, error) {109 func (s *UserStore) ListUsers(ctx context.Context) ([]*User, error) {
91 rows, err := s.db.QueryContext(ctx, `110 rows, err := s.db.QueryContext(ctx, `
92 SELECT did, indexed_at, updated_at, follows_dirty111 SELECT did, indexed_at, updated_at, follows_dirty
modified main.go +33 -3
@@ -33,6 +33,7 @@ func main() {
3333 fetchInterval := flag.Duration("fetch-interval", envDuration("GLEAN_FETCH_INTERVAL", 15*time.Minute), "feed fetch tick interval")
3434 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
3535 backfillConcurrency := flag.Int("backfill-concurrency", envInt("GLEAN_BACKFILL_CONCURRENCY", 5), "max concurrent backfill workers")
36+ articleRetentionDays := flag.Int("article-retention-days", envInt("GLEAN_ARTICLE_RETENTION_DAYS", 30), "delete articles older than this many days")
3637 sessionKey := envOr("GLEAN_SESSION_KEY", "")
3738 flag.Parse()
3839
@@ -110,11 +111,40 @@ func main() {
110111 fetcher := feed.NewFetcher(siteFetcher)
111112 srv := server.New(dbs, clientID, frontendURL, scheduler, fetcher, engine, logger, []byte(sessionKey), llm)
112113
113- cron := cluster.NewCron(engine, *clusterInterval, logger, dbs)
114+ cron := cluster.NewCron(engine, *clusterInterval, logger, dbs, *articleRetentionDays)
114115
115116 handler := atproto.NewStreamDBHandler(dbs.Articles, dbs.Users, logger)
116- jetstream := atproto.NewJetstreamConsumer(*jetstreamURL, handler.Handle, logger, dbs.CursorStore())
117-
117+ // Only stream events of known users; without this filter jetstream
118+ // delivers every matching record on the network and the database grows
119+ // unbounded.
120+ jetstream := atproto.NewJetstreamConsumer(*jetstreamURL, handler.Handle, logger, dbs.CursorStore(), dbs.Users.UserDIDList)
121+
122+ // Purge and reclaim disk space before the streaming jobs start writing
123+ // again: once the volume is full every SQLite write fails, including
124+ // OAuth sign-in.
125+ retentionCtx, cancelRetention := context.WithTimeout(context.Background(), 30*time.Minute)
126+ start := time.Now()
127+ unknownStats, err := dbs.PurgeUnknownUserRows(retentionCtx)
128+ if err != nil {
129+ logger.Error("initial purge of unknown-user rows failed", "error", err)
130+ } else if unknownStats.Total() > 0 {
131+ logger.Info("purged rows of unknown users", "stats", unknownStats)
132+ }
133+ expired, err := dbs.PurgeExpiredArticles(retentionCtx, *articleRetentionDays)
134+ if err != nil {
135+ logger.Error("initial article retention purge failed", "error", err)
136+ } else if expired > 0 {
137+ logger.Info("purged expired articles", "count", expired)
138+ }
139+ if unknownStats.Total()+expired > 0 {
140+ if err := dbs.ReclaimSpace(retentionCtx); err != nil {
141+ logger.Error("reclaiming database space incomplete", "error", err)
142+ } else {
143+ logger.Info("reclaimed database space")
144+ }
145+ }
146+ cancelRetention()
147+ logger.Info("initial retention complete", "elapsed", time.Since(start).Round(time.Second))
118148 ctx, cancel := context.WithCancel(context.Background())
119149 defer cancel()
120150
@@ -33,6 +33,7 @@ func main() {
33 fetchInterval := flag.Duration("fetch-interval", envDuration("GLEAN_FETCH_INTERVAL", 15*time.Minute), "feed fetch tick interval")33 fetchInterval := flag.Duration("fetch-interval", envDuration("GLEAN_FETCH_INTERVAL", 15*time.Minute), "feed fetch tick interval")
34 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")34 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
35 backfillConcurrency := flag.Int("backfill-concurrency", envInt("GLEAN_BACKFILL_CONCURRENCY", 5), "max concurrent backfill workers")35 backfillConcurrency := flag.Int("backfill-concurrency", envInt("GLEAN_BACKFILL_CONCURRENCY", 5), "max concurrent backfill workers")
36+ articleRetentionDays := flag.Int("article-retention-days", envInt("GLEAN_ARTICLE_RETENTION_DAYS", 30), "delete articles older than this many days")
36 sessionKey := envOr("GLEAN_SESSION_KEY", "")37 sessionKey := envOr("GLEAN_SESSION_KEY", "")
37 flag.Parse()38 flag.Parse()
38 39
@@ -110,11 +111,40 @@ func main() {
110 fetcher := feed.NewFetcher(siteFetcher)111 fetcher := feed.NewFetcher(siteFetcher)
111 srv := server.New(dbs, clientID, frontendURL, scheduler, fetcher, engine, logger, []byte(sessionKey), llm)112 srv := server.New(dbs, clientID, frontendURL, scheduler, fetcher, engine, logger, []byte(sessionKey), llm)
112 113
113- cron := cluster.NewCron(engine, *clusterInterval, logger, dbs)114+ cron := cluster.NewCron(engine, *clusterInterval, logger, dbs, *articleRetentionDays)
114 115
115 handler := atproto.NewStreamDBHandler(dbs.Articles, dbs.Users, logger)116 handler := atproto.NewStreamDBHandler(dbs.Articles, dbs.Users, logger)
116- jetstream := atproto.NewJetstreamConsumer(*jetstreamURL, handler.Handle, logger, dbs.CursorStore())117+ // Only stream events of known users; without this filter jetstream
117-118+ // delivers every matching record on the network and the database grows
119+ // unbounded.
120+ jetstream := atproto.NewJetstreamConsumer(*jetstreamURL, handler.Handle, logger, dbs.CursorStore(), dbs.Users.UserDIDList)
121+
122+ // Purge and reclaim disk space before the streaming jobs start writing
123+ // again: once the volume is full every SQLite write fails, including
124+ // OAuth sign-in.
125+ retentionCtx, cancelRetention := context.WithTimeout(context.Background(), 30*time.Minute)
126+ start := time.Now()
127+ unknownStats, err := dbs.PurgeUnknownUserRows(retentionCtx)
128+ if err != nil {
129+ logger.Error("initial purge of unknown-user rows failed", "error", err)
130+ } else if unknownStats.Total() > 0 {
131+ logger.Info("purged rows of unknown users", "stats", unknownStats)
132+ }
133+ expired, err := dbs.PurgeExpiredArticles(retentionCtx, *articleRetentionDays)
134+ if err != nil {
135+ logger.Error("initial article retention purge failed", "error", err)
136+ } else if expired > 0 {
137+ logger.Info("purged expired articles", "count", expired)
138+ }
139+ if unknownStats.Total()+expired > 0 {
140+ if err := dbs.ReclaimSpace(retentionCtx); err != nil {
141+ logger.Error("reclaiming database space incomplete", "error", err)
142+ } else {
143+ logger.Info("reclaimed database space")
144+ }
145+ }
146+ cancelRetention()
147+ logger.Info("initial retention complete", "elapsed", time.Since(start).Round(time.Second))
118 ctx, cancel := context.WithCancel(context.Background())148 ctx, cancel := context.WithCancel(context.Background())
119 defer cancel()149 defer cancel()
120 150
modified readme.md +1 -0
@@ -74,6 +74,7 @@ Then open `http://localhost:3000`.
7474 | `GLEAN_FETCH_INTERVAL` | `15m` | Feed fetch scheduler tick interval (Go duration) |
7575 | `GLEAN_COLLECTION_DIR_URL` | _(empty)_ | Collection directory URL for startup backfill |
7676 | `GLEAN_BACKFILL_CONCURRENCY` | `5` | Max concurrent backfill workers |
77+| `GLEAN_ARTICLE_RETENTION_DAYS` | `30` | Delete articles older than this many days; jetstream only streams events of known users |
7778 | `GLEAN_PLC_URL` | `https://plc.eurosky.network` | PLC directory URL for DID resolution |
7879 | `GLEAN_OAUTH_CLIENT_ID` | _(empty)_ | OAuth client-metadata URL; enables production OAuth (leave empty for localhost dev). Must resolve to this server's `/api/oauth/client-metadata` |
7980 | `GLEAN_FRONTEND_URL` | _(required)_ | Public origin of the SvelteKit frontend (e.g. `https://glean.at`); `make dev` defaults this to `http://localhost:3000` |
@@ -74,6 +74,7 @@ Then open `http://localhost:3000`.
74 | `GLEAN_FETCH_INTERVAL` | `15m` | Feed fetch scheduler tick interval (Go duration) |74 | `GLEAN_FETCH_INTERVAL` | `15m` | Feed fetch scheduler tick interval (Go duration) |
75 | `GLEAN_COLLECTION_DIR_URL` | _(empty)_ | Collection directory URL for startup backfill |75 | `GLEAN_COLLECTION_DIR_URL` | _(empty)_ | Collection directory URL for startup backfill |
76 | `GLEAN_BACKFILL_CONCURRENCY` | `5` | Max concurrent backfill workers |76 | `GLEAN_BACKFILL_CONCURRENCY` | `5` | Max concurrent backfill workers |
77+| `GLEAN_ARTICLE_RETENTION_DAYS` | `30` | Delete articles older than this many days; jetstream only streams events of known users |
77 | `GLEAN_PLC_URL` | `https://plc.eurosky.network` | PLC directory URL for DID resolution |78 | `GLEAN_PLC_URL` | `https://plc.eurosky.network` | PLC directory URL for DID resolution |
78 | `GLEAN_OAUTH_CLIENT_ID` | _(empty)_ | OAuth client-metadata URL; enables production OAuth (leave empty for localhost dev). Must resolve to this server's `/api/oauth/client-metadata` |79 | `GLEAN_OAUTH_CLIENT_ID` | _(empty)_ | OAuth client-metadata URL; enables production OAuth (leave empty for localhost dev). Must resolve to this server's `/api/oauth/client-metadata` |
79 | `GLEAN_FRONTEND_URL` | _(required)_ | Public origin of the SvelteKit frontend (e.g. `https://glean.at`); `make dev` defaults this to `http://localhost:3000` |80 | `GLEAN_FRONTEND_URL` | _(required)_ | Public origin of the SvelteKit frontend (e.g. `https://glean.at`); `make dev` defaults this to `http://localhost:3000` |