nandi/glean · 811355a

modified docs/specs.md +4 -1 Viewed

@@ -654,8 +654,9 @@ CREATE TABLE follows (
654	654	PRIMARY KEY (user_did, target_did)
655	655	);
656	656
657		-CREATE INDEX idx_follows_user ON follows(user_did);
658	657	CREATE INDEX idx_follows_target ON follows(target_did);
	658	+CREATE INDEX idx_follows_uri ON follows(uri);
	659	+CREATE INDEX idx_follows_followed_at ON follows(followed_at);
659	660	```
660	661
661	662	### 6.9 OAuth Storage (`<base>_users`)
@@ -804,6 +805,8 @@ A background goroutine runs on a configurable schedule (`GLEAN_CLUSTER_INTERVAL`
804	805	6. Compute follow distances: Incremental BFS for dirty users (1-hop through 3-hop from `follows` table)
805	806	7. Compute signal profiles: Per-user category/tag/like summaries
806	807	8. Auto-dismiss stale: Dismiss items shown >=5 times over >5 days without action
	808	+9. Prune old impressions: Delete `recommendation_impressions` older than 90 days
	809	+10. DB maintenance: Run `PRAGMA incremental_vacuum` on all 3 databases (users, articles, recs) to reclaim freed pages. Incremental auto-vacuum is enabled via `PRAGMA auto_vacuum = INCREMENTAL` at connection time, so pages freed by impression pruning and other deletions are reclaimed each cycle.
807	810
808	811	Jetstream ingestion and record indexing happen in a separate persistent goroutine (the Jetstream consumer), not in the cron.
809	812

      PRIMARY KEY (user_did, target_did)
  );
--CREATE INDEX idx_follows_user ON follows(user_did);
  CREATE INDEX idx_follows_target ON follows(target_did);
++CREATE INDEX idx_follows_uri ON follows(uri);
++CREATE INDEX idx_follows_followed_at ON follows(followed_at);
  ```
  ### 6.9 OAuth Storage (`<base>_users`)
 . **Compute follow distances**: Incremental BFS for dirty users (1-hop through 3-hop from `follows` table)
 . **Compute signal profiles**: Per-user category/tag/like summaries
 . **Auto-dismiss stale**: Dismiss items shown >=5 times over >5 days without action
++9. **Prune old impressions**: Delete `recommendation_impressions` older than 90 days
++10. **DB maintenance**: Run `PRAGMA incremental_vacuum` on all 3 databases (users, articles, recs) to reclaim freed pages. Incremental auto-vacuum is enabled via `PRAGMA auto_vacuum = INCREMENTAL` at connection time, so pages freed by impression pruning and other deletions are reclaimed each cycle.
  Jetstream ingestion and record indexing happen in a separate persistent goroutine (the Jetstream consumer), not in the cron.

modified internal/cluster/cron.go +13 -2 Viewed

@@ -5,6 +5,7 @@ import (
5	5	"log/slog"
6	6	"time"
7	7
	8	+ "pkg.rbrt.fr/glean/internal/db"
8	9	"pkg.rbrt.fr/glean/internal/metrics"
9	10	)
10	11
@@ -14,11 +15,17 @@ type Cron struct {
14	15	engine *Engine
15	16	interval time.Duration
16	17	logger *slog.Logger
	18	+ dbs *db.Store
17	19	}
18	20
19	21	// NewCron creates a new cron runner with the given engine and interval.
20		-func NewCron(engine Engine, interval time.Duration, logger slog.Logger) *Cron {
21		- return &Cron{engine: engine, interval: interval, logger: logger}
	22	+func NewCron(engine Engine, interval time.Duration, logger slog.Logger, dbs db.Store) Cron {
	23	+ return &Cron{
	24	+ engine: engine,
	25	+ interval: interval,
	26	+ logger: logger,
	27	+ dbs: dbs,
	28	+ }
22	29	}
23	30
24	31	// Run starts the cron loop. It blocks until ctx is cancelled. Each tick runs
@@ -65,6 +72,10 @@ func (c *Cron) Run(ctx context.Context) error {
65	72	c.engine.mu.Unlock()
66	73	}
67	74
	75	+ if err := c.dbs.RunMaintenance(ctx, 90); err != nil {
	76	+ c.logger.Error("db maintenance failed", "error", err)
	77	+ }
	78	+
68	79	metrics.ClusterRuns.Inc()
69	80	metrics.ClusterDuration.Observe(time.Since(start).Seconds())
70	81	c.logger.Info("similarity computation complete", "next_run", c.interval)

  	"log/slog"
  	"time"
++	"pkg.rbrt.fr/glean/internal/db"
  	"pkg.rbrt.fr/glean/internal/metrics"
  )
  	engine   *Engine
  	interval time.Duration
  	logger   *slog.Logger
++	dbs      *db.Store
  }
  // NewCron creates a new cron runner with the given engine and interval.
--func NewCron(engine *Engine, interval time.Duration, logger *slog.Logger) *Cron {
++func NewCron(engine *Engine, interval time.Duration, logger *slog.Logger, dbs *db.Store) *Cron {
--	return &Cron{engine: engine, interval: interval, logger: logger}
++	return &Cron{
++		engine:   engine,
++		interval: interval,
++		logger:   logger,
++		dbs:      dbs,
++	}
  }
  // Run starts the cron loop. It blocks until ctx is cancelled. Each tick runs
  			c.engine.mu.Unlock()
  		}
++		if err := c.dbs.RunMaintenance(ctx, 90); err != nil {
++			c.logger.Error("db maintenance failed", "error", err)
++		}
++
  		metrics.ClusterRuns.Inc()
  		metrics.ClusterDuration.Observe(time.Since(start).Seconds())
  		c.logger.Info("similarity computation complete", "next_run", c.interval)

modified internal/db/db.go +15 -1 Viewed

@@ -48,6 +48,7 @@ func Open(basePath string) (*Store, error) {
48	48	`PRAGMA cache = shared`,
49	49	`PRAGMA temp_store = FILE`,
50	50	`PRAGMA mmap_size = 268435456`,
	51	+ `PRAGMA auto_vacuum = INCREMENTAL`,
51	52	} {
52	53	if _, err := conn.Exec(p, nil); err != nil {
53	54	return err
@@ -125,6 +126,20 @@ func Open(basePath string) (*Store, error) {
125	126	}, nil
126	127	}
127	128
	129	+func (s *Store) RunMaintenance(ctx context.Context, impressionMaxAgeDays int) error {
	130	+ 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 {
	132	+ return fmt.Errorf("prune impressions: %w", err)
	133	+ }
	134	+
	135	+ for _, schema := range []string{"main", "articles", "recs"} {
	136	+ 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)
	138	+ }
	139	+ }
	140	+ return nil
	141	+}
	142	+
128	143	func (s *Store) Close() error {
129	144	if s.db != nil {
130	145	_ = s.db.Close()
@@ -235,7 +250,6 @@ var usersSchema = []string{
235	250	PRIMARY KEY (account_did, session_id)
236	251	)`,
237	252
238		- `CREATE INDEX IF NOT EXISTS idx_follows_user ON follows(user_did)`,
239	253	`CREATE INDEX IF NOT EXISTS idx_follows_target ON follows(target_did)`,
240	254	`CREATE INDEX IF NOT EXISTS idx_follows_uri ON follows(uri)`,
241	255	`CREATE INDEX IF NOT EXISTS idx_follows_followed_at ON follows(followed_at)`,

  				`PRAGMA cache = shared`,
  				`PRAGMA temp_store = FILE`,
  				`PRAGMA mmap_size = 268435456`,
++				`PRAGMA auto_vacuum = INCREMENTAL`,
  			} {
  				if _, err := conn.Exec(p, nil); err != nil {
  					return err
  	}, nil
  }
++func (s *Store) RunMaintenance(ctx context.Context, impressionMaxAgeDays int) error {
++	cutoff := time.Now().AddDate(0, 0, -impressionMaxAgeDays).Format(time.RFC3339)
++	if _, err := s.db.ExecContext(ctx, `DELETE FROM main.recommendation_impressions WHERE first_shown_at < ?`, cutoff); err != nil {
++		return fmt.Errorf("prune impressions: %w", err)
++	}
++
++	for _, schema := range []string{"main", "articles", "recs"} {
++		if _, err := s.db.ExecContext(ctx, fmt.Sprintf("PRAGMA %s.incremental_vacuum", schema)); err != nil {
++			return fmt.Errorf("incremental_vacuum %s: %w", schema, err)
++		}
++	}
++	return nil
++}
++
  func (s *Store) Close() error {
  	if s.db != nil {
  		_ = s.db.Close()
  		PRIMARY KEY (account_did, session_id)
  	)`,
--	`CREATE INDEX IF NOT EXISTS idx_follows_user ON follows(user_did)`,
  	`CREATE INDEX IF NOT EXISTS idx_follows_target ON follows(target_did)`,
  	`CREATE INDEX IF NOT EXISTS idx_follows_uri ON follows(uri)`,
  	`CREATE INDEX IF NOT EXISTS idx_follows_followed_at ON follows(followed_at)`,

modified internal/db/migrations.go +19 -1 Viewed

@@ -13,7 +13,7 @@ func init() {
13	13	}
14	14
15	15	// SchemaVersion must be incremented each time a migration is added to the migrations slice (used so that fresh dbs skip running migrations).
16		-const SchemaVersion = 5
	16	+const SchemaVersion = 6
17	17
18	18	type migration struct {
19	19	id int
@@ -47,6 +47,11 @@ var migrations = []migration{
47	47	name: "user_settings_expanded_view",
48	48	run: migrateUserSettingsExpandedView,
49	49	},
	50	+ {
	51	+ id: 6,
	52	+ name: "drop_redundant_follows_user_index",
	53	+ run: migrateDropFollowsUserIndex,
	54	+ },
50	55	}
51	56
52	57	func runMigrations(db *DB) error {
@@ -248,3 +253,16 @@ func migrateUserSettingsExpandedView(db *DB) error {
248	253	}
249	254	return nil
250	255	}
	256	+
	257	+func migrateDropFollowsUserIndex(db *DB) error {
	258	+ var name string
	259	+ err := db.QueryRow("SELECT name FROM sqlite_master WHERE type='index' AND name='idx_follows_user'").Scan(&name)
	260	+ if err == sql.ErrNoRows {
	261	+ return nil
	262	+ }
	263	+ if err != nil {
	264	+ return fmt.Errorf("check idx_follows_user: %w", err)
	265	+ }
	266	+ _, err = db.Exec("DROP INDEX IF EXISTS idx_follows_user")
	267	+ return err
	268	+}

  }
  // SchemaVersion must be incremented each time a migration is added to the migrations slice (used so that fresh dbs skip running migrations).
--const SchemaVersion = 5
++const SchemaVersion = 6
  type migration struct {
  	id   int
  		name: "user_settings_expanded_view",
  		run:  migrateUserSettingsExpandedView,
  	},
++	{
++		id:   6,
++		name: "drop_redundant_follows_user_index",
++		run:  migrateDropFollowsUserIndex,
++	},
  }
  func runMigrations(db *DB) error {
  	}
  	return nil
  }
++
++func migrateDropFollowsUserIndex(db *DB) error {
++	var name string
++	err := db.QueryRow("SELECT name FROM sqlite_master WHERE type='index' AND name='idx_follows_user'").Scan(&name)
++	if err == sql.ErrNoRows {
++		return nil
++	}
++	if err != nil {
++		return fmt.Errorf("check idx_follows_user: %w", err)
++	}
++	_, err = db.Exec("DROP INDEX IF EXISTS idx_follows_user")
++	return err
++}

modified internal/feedback/feedback.go +13 -0 Viewed

@@ -92,3 +92,16 @@ func (s *Service) IsFeedDismissed(ctx context.Context, userDID, feedURL string)
92	92	`, userDID, feedURL).Scan(&count)
93	93	return count > 0, err
94	94	}
	95	+
	96	+func (s *Service) PruneOldImpressions(ctx context.Context, maxAgeDays int) (int64, error) {
	97	+ cutoff := time.Now().AddDate(0, 0, -maxAgeDays).Format(time.RFC3339)
	98	+
	99	+ res, err := s.db.ExecContext(ctx, `
	100	+ DELETE FROM main.recommendation_impressions
	101	+ WHERE first_shown_at < ?
	102	+ `, cutoff)
	103	+ if err != nil {
	104	+ return 0, err
	105	+ }
	106	+ return res.RowsAffected()
	107	+}

  	`, userDID, feedURL).Scan(&count)
  	return count > 0, err
  }
++
++func (s *Service) PruneOldImpressions(ctx context.Context, maxAgeDays int) (int64, error) {
++	cutoff := time.Now().AddDate(0, 0, -maxAgeDays).Format(time.RFC3339)
++
++	res, err := s.db.ExecContext(ctx, `
++		DELETE FROM main.recommendation_impressions
++		WHERE first_shown_at < ?
++	`, cutoff)
++	if err != nil {
++		return 0, err
++	}
++	return res.RowsAffected()
++}

modified main.go +1 -1 Viewed

@@ -106,7 +106,7 @@ func main() {
106	106	fetcher := feed.NewFetcher(siteFetcher)
107	107	srv := server.New(dbs, clientID, callbackURL, *addr, scheduler, fetcher, engine, logger, []byte(sessionKey))
108	108
109		- cron := cluster.NewCron(engine, *clusterInterval, logger)
	109	+ cron := cluster.NewCron(engine, *clusterInterval, logger, dbs)
110	110
111	111	handler := atproto.NewStreamDBHandler(dbs.Articles, dbs.Users, logger)
112	112	jetstream := atproto.NewJetstreamConsumer(*jetstreamURL, handler.Handle, logger, dbs.CursorStore())

  	fetcher := feed.NewFetcher(siteFetcher)
  	srv := server.New(dbs, clientID, callbackURL, *addr, scheduler, fetcher, engine, logger, []byte(sessionKey))
--	cron := cluster.NewCron(engine, *clusterInterval, logger)
++	cron := cluster.NewCron(engine, *clusterInterval, logger, dbs)
  	handler := atproto.NewStreamDBHandler(dbs.Articles, dbs.Users, logger)
  	jetstream := atproto.NewJetstreamConsumer(*jetstreamURL, handler.Handle, logger, dbs.CursorStore())

Add cron-based database maintenance and cleanup indexesUnverified