Commit 6a6dbc43 authored by Steven Allen's avatar Steven Allen

fix: ignore bad peers when fetching the latest value

They may not support the protocol. Or they may just not give us what we're
looking for. We might as well keep asking till someone answers.
parent eebeec43
...@@ -478,7 +478,7 @@ func (p *PubsubValueStore) handleSubscription(ctx context.Context, ti *topicInfo ...@@ -478,7 +478,7 @@ func (p *PubsubValueStore) handleSubscription(ctx context.Context, ti *topicInfo
case <-ctx.Done(): case <-ctx.Done():
return return
default: default:
log.Errorf("PubsubPeerJoin: error interacting with new peer", err) log.Errorf("PubsubPeerJoin: error interacting with new peer: %s", err)
} }
} }
} }
...@@ -524,15 +524,7 @@ func (p *PubsubValueStore) handleNewMsgs(ctx context.Context, sub *pubsub.Subscr ...@@ -524,15 +524,7 @@ func (p *PubsubValueStore) handleNewMsgs(ctx context.Context, sub *pubsub.Subscr
} }
func (p *PubsubValueStore) handleNewPeer(ctx context.Context, peerEvtHandler *pubsub.TopicEventHandler, key string) ([]byte, error) { func (p *PubsubValueStore) handleNewPeer(ctx context.Context, peerEvtHandler *pubsub.TopicEventHandler, key string) ([]byte, error) {
select { for ctx.Err() == nil {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
var pid peer.ID
for {
peerEvt, err := peerEvtHandler.NextPeerEvent(ctx) peerEvt, err := peerEvtHandler.NextPeerEvent(ctx)
if err != nil { if err != nil {
if err != context.Canceled { if err != context.Canceled {
...@@ -540,13 +532,19 @@ func (p *PubsubValueStore) handleNewPeer(ctx context.Context, peerEvtHandler *pu ...@@ -540,13 +532,19 @@ func (p *PubsubValueStore) handleNewPeer(ctx context.Context, peerEvtHandler *pu
} }
return nil, err return nil, err
} }
if peerEvt.Type == pubsub.PeerJoin {
pid = peerEvt.Peer if peerEvt.Type != pubsub.PeerJoin {
break continue
} }
}
return p.fetch.Fetch(ctx, pid, key) pid := peerEvt.Peer
value, err := p.fetch.Fetch(ctx, pid, key)
if err == nil {
return value, nil
}
log.Debugf("failed to fetch latest pubsub value for key '%s' from peer '%s': %s", key, pid, err)
}
return nil, ctx.Err()
} }
func (p *PubsubValueStore) notifyWatchers(key string, data []byte) { func (p *PubsubValueStore) notifyWatchers(key string, data []byte) {
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment