mirror of
https://github.com/nbd-wtf/go-nostr.git
synced 2025-09-05 20:50:34 +02:00
fix atrocious bug on pool.subMany(): we were missing events because of a badly designed select{}
This commit is contained in:
40
pool.go
40
pool.go
@@ -72,14 +72,15 @@ func (pool *SimplePool) SubManyNonUnique(ctx context.Context, urls []string, fil
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filters, unique bool) chan IncomingEvent {
|
func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filters, unique bool) chan IncomingEvent {
|
||||||
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
|
_ = cancel // do this so `go vet` will stop complaining
|
||||||
events := make(chan IncomingEvent)
|
events := make(chan IncomingEvent)
|
||||||
seenAlready := xsync.NewMapOf[Timestamp]()
|
seenAlready := xsync.NewMapOf[Timestamp]()
|
||||||
ticker := time.NewTicker(seenAlreadyDropTick)
|
ticker := time.NewTicker(seenAlreadyDropTick)
|
||||||
eose := false
|
eose := false
|
||||||
|
|
||||||
pending := xsync.NewCounter()
|
pending := xsync.NewCounter()
|
||||||
initial := len(urls)
|
pending.Add(int64(len(urls)))
|
||||||
pending.Add(int64(initial))
|
|
||||||
for _, url := range urls {
|
for _, url := range urls {
|
||||||
go func(nm string) {
|
go func(nm string) {
|
||||||
relay, err := pool.EnsureRelay(nm)
|
relay, err := pool.EnsureRelay(nm)
|
||||||
@@ -92,23 +93,38 @@ func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filt
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
for evt := range sub.Events {
|
defer func() {
|
||||||
if unique {
|
pending.Dec()
|
||||||
if _, seen := seenAlready.LoadOrStore(evt.ID, evt.CreatedAt); seen {
|
if pending.Value() == 0 {
|
||||||
continue
|
close(events)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
cancel()
|
||||||
|
}()
|
||||||
|
|
||||||
|
for {
|
||||||
select {
|
select {
|
||||||
|
case evt, more := <-sub.Events:
|
||||||
|
if !more {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if unique {
|
||||||
|
if _, seen := seenAlready.LoadOrStore(evt.ID, evt.CreatedAt); seen {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case events <- IncomingEvent{Event: evt, Relay: relay}:
|
||||||
|
case <-ctx.Done():
|
||||||
|
}
|
||||||
case <-sub.EndOfStoredEvents:
|
case <-sub.EndOfStoredEvents:
|
||||||
eose = true
|
eose = true
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
if eose {
|
if eose {
|
||||||
del := map[string]struct{}{}
|
del := map[string]struct{}{}
|
||||||
old := Timestamp(time.Now().Add(-seenAlreadyDropTick).Unix())
|
old := Timestamp(time.Now().Add(-seenAlreadyDropTick).Unix())
|
||||||
seenAlready.Range(func(key string, value Timestamp) bool {
|
seenAlready.Range(func(id string, value Timestamp) bool {
|
||||||
if value < old {
|
if value < old {
|
||||||
del[evt.ID] = struct{}{}
|
del[id] = struct{}{}
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
})
|
})
|
||||||
@@ -116,7 +132,6 @@ func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filt
|
|||||||
seenAlready.Delete(k)
|
seenAlready.Delete(k)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
case events <- IncomingEvent{Event: evt, Relay: relay}:
|
|
||||||
case reason := <-sub.ClosedReason:
|
case reason := <-sub.ClosedReason:
|
||||||
log.Printf("CLOSED from %s: '%s'\n", nm, reason)
|
log.Printf("CLOSED from %s: '%s'\n", nm, reason)
|
||||||
return
|
return
|
||||||
@@ -124,11 +139,6 @@ func (pool *SimplePool) subMany(ctx context.Context, urls []string, filters Filt
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pending.Dec()
|
|
||||||
if pending.Value() == 0 {
|
|
||||||
close(events)
|
|
||||||
}
|
|
||||||
}(NormalizeURL(url))
|
}(NormalizeURL(url))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user