rickub
nandi/gleanpublic⑂ Fork 0
⑂ a19fa48
Branches
claude/modal-container-min-76527a 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.

Add configurable concurrency to collection directory backfillUnverified

Julien Robert committed 2026-04-22T02:29:16+02:00 Browse files
a19fa48 parent: b8b7aa8
modified internal/server/server.go +34 -21
@@ -11,6 +11,7 @@ import (
1111 "slices"
1212 "strconv"
1313 "strings"
14+ "sync"
1415 "time"
1516
1617 "github.com/go-chi/chi/v5"
@@ -441,7 +442,7 @@ func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {
441442 }
442443 }
443444
444-func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL string) {
445+func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL string, concurrency int) {
445446 if collectionDirURL == "" {
446447 return
447448 }
@@ -469,36 +470,48 @@ func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL
469470
470471 s.logger.Info("collection directory backfill", "total", len(dids), "missing", len(missing))
471472
473+ sem := make(chan struct{}, concurrency)
474+ var wg sync.WaitGroup
475+
472476 for _, did := range missing {
473477 if ctx.Err() != nil {
474- return
478+ break
475479 }
476480
477- handle := did
478- if ident, err := atproto.ResolveIdentity(ctx, did); err == nil {
479- handle = ident.Handle.String()
480- }
481+ sem <- struct{}{}
482+ wg.Add(1)
481483
482- if _, err := s.db.CreateUser(ctx, did, handle, "", ""); err != nil {
483- s.logger.Error("failed to create user during backfill", "error", err, "did", did)
484- continue
485- }
484+ go func(did string) {
485+ defer func() { <-sem }()
486+ defer wg.Done()
486487
487- pdsURL, err := atproto.ResolvePDSEndpoint(ctx, did)
488- if err != nil {
489- s.logger.Error("failed to resolve PDS for backfill", "error", err, "did", did)
490- continue
491- }
488+ handle := did
489+ if ident, err := atproto.ResolveIdentity(ctx, did); err == nil {
490+ handle = ident.Handle.String()
491+ }
492492
493- client := atproto.NewUnauthenticatedClient(pdsURL)
494- sync := atproto.NewSync(s.db, client, s.logger)
495- if err := sync.Run(ctx, did); err != nil {
496- s.logger.Error("backfill sync failed", "error", err, "did", did)
497- }
493+ if _, err := s.db.CreateUser(ctx, did, handle, "", ""); err != nil {
494+ s.logger.Error("failed to create user during backfill", "error", err, "did", did)
495+ return
496+ }
497+
498+ pdsURL, err := atproto.ResolvePDSEndpoint(ctx, did)
499+ if err != nil {
500+ s.logger.Error("failed to resolve PDS for backfill", "error", err, "did", did)
501+ return
502+ }
503+
504+ client := atproto.NewUnauthenticatedClient(pdsURL)
505+ sync := atproto.NewSync(s.db, client, s.logger)
506+ if err := sync.Run(ctx, did); err != nil {
507+ s.logger.Error("backfill sync failed", "error", err, "did", did)
508+ }
498509
499- s.refreshUserFeeds(ctx, did)
510+ s.refreshUserFeeds(ctx, did)
511+ }(did)
500512 }
501513
514+ wg.Wait()
502515 s.logger.Info("collection directory backfill complete")
503516 }
504517
@@ -11,6 +11,7 @@ import (
11 "slices"11 "slices"
12 "strconv"12 "strconv"
13 "strings"13 "strings"
14+ "sync"
14 "time"15 "time"
15 16
16 "github.com/go-chi/chi/v5"17 "github.com/go-chi/chi/v5"
@@ -441,7 +442,7 @@ func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {
441 }442 }
442 }443 }
443 444
444-func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL string) {445+func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL string, concurrency int) {
445 if collectionDirURL == "" {446 if collectionDirURL == "" {
446 return447 return
447 }448 }
@@ -469,36 +470,48 @@ func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL
469 470
470 s.logger.Info("collection directory backfill", "total", len(dids), "missing", len(missing))471 s.logger.Info("collection directory backfill", "total", len(dids), "missing", len(missing))
471 472
473+ sem := make(chan struct{}, concurrency)
474+ var wg sync.WaitGroup
475+
472 for _, did := range missing {476 for _, did := range missing {
473 if ctx.Err() != nil {477 if ctx.Err() != nil {
474- return478+ break
475 }479 }
476 480
477- handle := did481+ sem <- struct{}{}
478- if ident, err := atproto.ResolveIdentity(ctx, did); err == nil {482+ wg.Add(1)
479- handle = ident.Handle.String()
480- }
481 483
482- if _, err := s.db.CreateUser(ctx, did, handle, "", ""); err != nil {484+ go func(did string) {
483- s.logger.Error("failed to create user during backfill", "error", err, "did", did)485+ defer func() { <-sem }()
484- continue486+ defer wg.Done()
485- }
486 487
487- pdsURL, err := atproto.ResolvePDSEndpoint(ctx, did)488+ handle := did
488- if err != nil {489+ if ident, err := atproto.ResolveIdentity(ctx, did); err == nil {
489- s.logger.Error("failed to resolve PDS for backfill", "error", err, "did", did)490+ handle = ident.Handle.String()
490- continue491+ }
491- }
492 492
493- client := atproto.NewUnauthenticatedClient(pdsURL)493+ if _, err := s.db.CreateUser(ctx, did, handle, "", ""); err != nil {
494- sync := atproto.NewSync(s.db, client, s.logger)494+ s.logger.Error("failed to create user during backfill", "error", err, "did", did)
495- if err := sync.Run(ctx, did); err != nil {495+ return
496- s.logger.Error("backfill sync failed", "error", err, "did", did)496+ }
497- }497+
498+ pdsURL, err := atproto.ResolvePDSEndpoint(ctx, did)
499+ if err != nil {
500+ s.logger.Error("failed to resolve PDS for backfill", "error", err, "did", did)
501+ return
502+ }
503+
504+ client := atproto.NewUnauthenticatedClient(pdsURL)
505+ sync := atproto.NewSync(s.db, client, s.logger)
506+ if err := sync.Run(ctx, did); err != nil {
507+ s.logger.Error("backfill sync failed", "error", err, "did", did)
508+ }
498 509
499- s.refreshUserFeeds(ctx, did)510+ s.refreshUserFeeds(ctx, did)
511+ }(did)
500 }512 }
501 513
514+ wg.Wait()
502 s.logger.Info("collection directory backfill complete")515 s.logger.Info("collection directory backfill complete")
503 }516 }
504 517
modified main.go +12 -1
@@ -8,6 +8,7 @@ import (
88 "net/http"
99 "os"
1010 "os/signal"
11+ "strconv"
1112 "syscall"
1213 "time"
1314
@@ -25,6 +26,7 @@ func main() {
2526 syncInterval := flag.Duration("sync-interval", envDuration("GLEAN_SYNC_INTERVAL", 1*time.Hour), "PDS sync interval")
2627 clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 10*time.Minute), "cluster recomputation interval")
2728 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
29+ backfillConcurrency := flag.Int("backfill-concurrency", envInt("GLEAN_BACKFILL_CONCURRENCY", 5), "max concurrent backfill workers")
2830 flag.Parse()
2931
3032 atproto.InitIdentity(envOr("GLEAN_PLC_URL", "https://didplc.glean.at"))
@@ -70,7 +72,7 @@ func main() {
7072 srv.PeriodicSync(ctx, *syncInterval)
7173 }()
7274 go func() {
73- srv.BackfillFromCollectionDir(ctx, *collectionDirURL)
75+ srv.BackfillFromCollectionDir(ctx, *collectionDirURL, *backfillConcurrency)
7476 }()
7577 go func() {
7678 if err := jetstream.Start(ctx); err != nil && ctx.Err() == nil {
@@ -125,3 +127,12 @@ func envDuration(key string, fallback time.Duration) time.Duration {
125127 }
126128 return fallback
127129 }
130+
131+func envInt(key string, fallback int) int {
132+ if v := os.Getenv(key); v != "" {
133+ if n, err := strconv.Atoi(v); err == nil {
134+ return n
135+ }
136+ }
137+ return fallback
138+}
@@ -8,6 +8,7 @@ import (
8 "net/http"8 "net/http"
9 "os"9 "os"
10 "os/signal"10 "os/signal"
11+ "strconv"
11 "syscall"12 "syscall"
12 "time"13 "time"
13 14
@@ -25,6 +26,7 @@ func main() {
25 syncInterval := flag.Duration("sync-interval", envDuration("GLEAN_SYNC_INTERVAL", 1*time.Hour), "PDS sync interval")26 syncInterval := flag.Duration("sync-interval", envDuration("GLEAN_SYNC_INTERVAL", 1*time.Hour), "PDS sync interval")
26 clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 10*time.Minute), "cluster recomputation interval")27 clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 10*time.Minute), "cluster recomputation interval")
27 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")28 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
29+ backfillConcurrency := flag.Int("backfill-concurrency", envInt("GLEAN_BACKFILL_CONCURRENCY", 5), "max concurrent backfill workers")
28 flag.Parse()30 flag.Parse()
29 31
30 atproto.InitIdentity(envOr("GLEAN_PLC_URL", "https://didplc.glean.at"))32 atproto.InitIdentity(envOr("GLEAN_PLC_URL", "https://didplc.glean.at"))
@@ -70,7 +72,7 @@ func main() {
70 srv.PeriodicSync(ctx, *syncInterval)72 srv.PeriodicSync(ctx, *syncInterval)
71 }()73 }()
72 go func() {74 go func() {
73- srv.BackfillFromCollectionDir(ctx, *collectionDirURL)75+ srv.BackfillFromCollectionDir(ctx, *collectionDirURL, *backfillConcurrency)
74 }()76 }()
75 go func() {77 go func() {
76 if err := jetstream.Start(ctx); err != nil && ctx.Err() == nil {78 if err := jetstream.Start(ctx); err != nil && ctx.Err() == nil {
@@ -125,3 +127,12 @@ func envDuration(key string, fallback time.Duration) time.Duration {
125 }127 }
126 return fallback128 return fallback
127 }129 }
130+
131+func envInt(key string, fallback int) int {
132+ if v := os.Getenv(key); v != "" {
133+ if n, err := strconv.Atoi(v); err == nil {
134+ return n
135+ }
136+ }
137+ return fallback
138+}