| 798 | } |
| 799 | |
| 800 | func (m *mapper) bestToUpdate(best map[uuid.UUID]mapping) *proto.CoordinateResponse { |
| 801 | resp := new(proto.CoordinateResponse) |
| 802 | |
| 803 | for k, mpng := range best { |
| 804 | var reason string |
| 805 | sm, ok := m.sent[k] |
| 806 | switch { |
| 807 | case !ok && mpng.kind == proto.CoordinateResponse_PeerUpdate_LOST: |
| 808 | // we don't need to send a "lost" update if we've never sent an update about this peer |
| 809 | continue |
| 810 | case !ok && mpng.kind == proto.CoordinateResponse_PeerUpdate_NODE: |
| 811 | reason = "new" |
| 812 | case ok && sm.kind == proto.CoordinateResponse_PeerUpdate_LOST && mpng.kind == proto.CoordinateResponse_PeerUpdate_LOST: |
| 813 | // was lost and remains lost, no update needed |
| 814 | continue |
| 815 | case ok && sm.kind == proto.CoordinateResponse_PeerUpdate_LOST && mpng.kind == proto.CoordinateResponse_PeerUpdate_NODE: |
| 816 | reason = "found" |
| 817 | case ok && sm.kind == proto.CoordinateResponse_PeerUpdate_NODE && mpng.kind == proto.CoordinateResponse_PeerUpdate_LOST: |
| 818 | reason = "lost" |
| 819 | case ok && sm.kind == proto.CoordinateResponse_PeerUpdate_NODE && mpng.kind == proto.CoordinateResponse_PeerUpdate_NODE: |
| 820 | eq, err := sm.node.Equal(mpng.node) |
| 821 | if err != nil { |
| 822 | m.logger.Critical(m.ctx, "failed to compare nodes", slog.F("old", sm.node), slog.F("new", mpng.node)) |
| 823 | continue |
| 824 | } |
| 825 | if eq { |
| 826 | continue |
| 827 | } |
| 828 | reason = "update" |
| 829 | } |
| 830 | resp.PeerUpdates = append(resp.PeerUpdates, &proto.CoordinateResponse_PeerUpdate{ |
| 831 | Id: agpl.UUIDToByteSlice(k), |
| 832 | Node: mpng.node, |
| 833 | Kind: mpng.kind, |
| 834 | Reason: reason, |
| 835 | }) |
| 836 | m.sent[k] = mpng |
| 837 | } |
| 838 | |
| 839 | for k := range m.sent { |
| 840 | if _, ok := best[k]; !ok { |
| 841 | resp.PeerUpdates = append(resp.PeerUpdates, &proto.CoordinateResponse_PeerUpdate{ |
| 842 | Id: agpl.UUIDToByteSlice(k), |
| 843 | Kind: proto.CoordinateResponse_PeerUpdate_DISCONNECTED, |
| 844 | Reason: "disconnected", |
| 845 | }) |
| 846 | delete(m.sent, k) |
| 847 | } |
| 848 | } |
| 849 | |
| 850 | if len(resp.PeerUpdates) == 0 { |
| 851 | return nil |
| 852 | } |
| 853 | return resp |
| 854 | } |
| 855 | |
| 856 | // querier is responsible for monitoring pubsub notifications and querying the database for the |
| 857 | // mappings that all connected peers need. It also checks heartbeats and withdraws mappings from |