nandi/glean · c8ebf4d

modified internal/atproto/sync.go +1 -1 Viewed

@@ -267,7 +267,7 @@ func (s *Sync) syncFollows(ctx context.Context, userDID string) error {
267	267	}
268	268
269	269	g, gCtx := errgroup.WithContext(ctx)
270		- g.SetLimit(10)
	270	+ g.SetLimit(20)
271	271
272	272	dids := make([]string, 0, len(activeFollows))
273	273	profiles := make([]db.UserData, len(activeFollows))

  	}
  	g, gCtx := errgroup.WithContext(ctx)
--	g.SetLimit(10)
++	g.SetLimit(20)
  	dids := make([]string, 0, len(activeFollows))
  	profiles := make([]db.UserData, len(activeFollows))

modified internal/db/multi.go +0 -8 Viewed

@@ -23,14 +23,6 @@ func OpenAll(basePath string) (*Databases, error) {
23	23	articlesPath := basePath + "_articles"
24	24	recsPath := basePath + "_recs"
25	25
26		- for _, p := range []string{articlesPath, recsPath} {
27		- f, err := sql.Open("sqlite3", p+"?"+DSN)
28		- if err != nil {
29		- return nil, err
30		- }
31		- f.Close()
32		- }
33		-
34	26	seq := atomic.AddInt64(&multiDriverSeq, 1)
35	27	driverName := fmt.Sprintf("sqlite3_glean_multi_%d", seq)
36	28

  	articlesPath := basePath + "_articles"
  	recsPath := basePath + "_recs"
--	for _, p := range []string{articlesPath, recsPath} {
--		f, err := sql.Open("sqlite3", p+"?"+DSN)
--		if err != nil {
--			return nil, err
--		}
--		f.Close()
--	}
--
  	seq := atomic.AddInt64(&multiDriverSeq, 1)
  	driverName := fmt.Sprintf("sqlite3_glean_multi_%d", seq)

modified internal/feed/fetcher.go +12 -14 Viewed

@@ -8,6 +8,7 @@ import (
8	8	"sync"
9	9	"time"
10	10
	11	+ "golang.org/x/sync/errgroup"
11	12	"pkg.rbrt.fr/glean/internal/httpclient"
12	13	"pkg.rbrt.fr/glean/internal/metrics"
13	14	)
@@ -171,26 +172,23 @@ func (s *Scheduler) Run(ctx context.Context) error {
171	172	}
172	173
173	174	func (s *Scheduler) fetchAll(ctx context.Context) {
174		- feeds, err := s.store.GetFeedsToFetch(ctx, s.staleInterval, 10_000)
	175	+ feeds, err := s.store.GetFeedsToFetch(ctx, s.staleInterval, 1_000)
175	176	if err != nil {
176	177	s.logger.Error("failed to get feeds", "error", err)
177	178	return
178	179	}
179	180
180		- sem := make(chan struct{}, 10)
181		- var wg sync.WaitGroup
	181	+ s.logger.Info("feeds fetched", "count", len(feeds))
	182	+
	183	+ g, gCtx := errgroup.WithContext(ctx)
	184	+ g.SetLimit(20)
182	185	for _, f := range feeds {
183		- wg.Add(1)
184		- sem <- struct{}{}
185		- go func(feed *Feed) {
186		- defer func() {
187		- <-sem
188		- wg.Done()
189		- }()
190		- s.FetchFeed(ctx, feed)
191		- }(f)
192		- }
193		- wg.Wait()
	186	+ g.Go(func() error {
	187	+ s.FetchFeed(gCtx, f)
	188	+ return nil
	189	+ })
	190	+ }
	191	+ _ = g.Wait()
194	192	}
195	193
196	194	func (s Scheduler) FetchFeed(ctx context.Context, feed Feed) {

  	"sync"
  	"time"
++	"golang.org/x/sync/errgroup"
  	"pkg.rbrt.fr/glean/internal/httpclient"
  	"pkg.rbrt.fr/glean/internal/metrics"
  )
  }
  func (s *Scheduler) fetchAll(ctx context.Context) {
--	feeds, err := s.store.GetFeedsToFetch(ctx, s.staleInterval, 10_000)
++	feeds, err := s.store.GetFeedsToFetch(ctx, s.staleInterval, 1_000)
  	if err != nil {
  		s.logger.Error("failed to get feeds", "error", err)
  		return
  	}
--	sem := make(chan struct{}, 10)
++	s.logger.Info("feeds fetched", "count", len(feeds))
--	var wg sync.WaitGroup
++
++	g, gCtx := errgroup.WithContext(ctx)
++	g.SetLimit(20)
  	for _, f := range feeds {
--		wg.Add(1)
++		g.Go(func() error {
--		sem <- struct{}{}
++			s.FetchFeed(gCtx, f)
--		go func(feed *Feed) {
++			return nil
--			defer func() {
++		})
--				<-sem
++	}
--				wg.Done()
++	_ = g.Wait()
--			}()
--			s.FetchFeed(ctx, feed)
--		}(f)
--	}
--	wg.Wait()
  }
  func (s *Scheduler) FetchFeed(ctx context.Context, feed *Feed) {

modified internal/httpclient/httpclient.go +3 -3 Viewed

@@ -19,8 +19,9 @@ const (
19	19	func NewTransport() *http.Transport {
20	20	return &http.Transport{
21	21	DialContext: (&net.Dialer{
22		- Timeout: 10 * time.Second,
23		- KeepAlive: 15 * time.Second,
	22	+ Timeout: 10 * time.Second,
	23	+ KeepAlive: 15 * time.Second,
	24	+ FallbackDelay: 300 * time.Millisecond,
24	25	}).DialContext,
25	26	MaxIdleConns: 50,
26	27	IdleConnTimeout: 10 * time.Second,
@@ -31,7 +32,6 @@ func NewTransport() *http.Transport {
31	32
32	33	func SetDefaultHeaders(req *http.Request) {
33	34	req.Header.Set("User-Agent", UserAgent)
34		- req.Header.Set("Accept-Encoding", "br,gzip")
35	35	req.Header.Set("Connection", "close")
36	36	}
37	37

  func NewTransport() *http.Transport {
  	return &http.Transport{
  		DialContext: (&net.Dialer{
--			Timeout:   10 * time.Second,
++			Timeout:         10 * time.Second,
--			KeepAlive: 15 * time.Second,
++			KeepAlive:       15 * time.Second,
++			FallbackDelay:   300 * time.Millisecond,
  		}).DialContext,
  		MaxIdleConns:        50,
  		IdleConnTimeout:     10 * time.Second,
  func SetDefaultHeaders(req *http.Request) {
  	req.Header.Set("User-Agent", UserAgent)
--	req.Header.Set("Accept-Encoding", "br,gzip")
  	req.Header.Set("Connection", "close")
  }

Optimize feed fetching and improve HTTP transportUnverified