V2 host scanning

explorer/explorer.go

@@ -67,7 +67,7 @@ type Store interface {
 	SiafundElements(ids []types.SiafundOutputID) (result []SiafundOutput, err error)
 
 	Hosts(pks []types.PublicKey) ([]Host, error)
-	HostsForScanning(maxLastScan, minLastAnnouncement time.Time, offset, limit uint64) ([]chain.HostAnnouncement, error)
+	HostsForScanning(maxLastScan, minLastAnnouncement time.Time, offset, limit uint64) ([]Host, error)
 }
 
 // Explorer implements a Sia explorer.

explorer/scan.go

@@ -9,7 +9,7 @@ import (
 
 	crhpv2 "go.sia.tech/core/rhp/v2"
 	"go.sia.tech/core/types"
-	"go.sia.tech/coreutils/chain"
+	crhpv4 "go.sia.tech/coreutils/rhp/v4"
 	"go.sia.tech/explored/internal/geoip"
 	rhpv2 "go.sia.tech/explored/internal/rhp/v2"
 	rhpv3 "go.sia.tech/explored/internal/rhp/v3"
@@ -52,11 +52,12 @@ func (e *Explorer) waitForSync() error {
 	return nil
 }
 
-func (e *Explorer) scanHost(locator geoip.Locator, host chain.HostAnnouncement) (HostScan, error) {
+func (e *Explorer) scanV1Host(locator geoip.Locator, host Host) (HostScan, error) {
 	ctx, cancel := context.WithTimeout(e.ctx, e.scanCfg.Timeout)
 	defer cancel()
 
 	dialer := (&net.Dialer{})
+
 	conn, err := dialer.DialContext(ctx, "tcp", host.NetAddress)
 	if err != nil {
 		return HostScan{}, fmt.Errorf("scanHost: failed to connect to host: %w", err)
@@ -80,7 +81,6 @@ func (e *Explorer) scanHost(locator geoip.Locator, host chain.HostAnnouncement)
 	}
 
 	resolved, err := net.ResolveIPAddr("ip", hostIP)
-	// if we can resolve the address
 	if err != nil {
 		return HostScan{}, fmt.Errorf("scanHost: failed to resolve host address: %w", err)
 	}
@@ -113,7 +113,53 @@ func (e *Explorer) scanHost(locator geoip.Locator, host chain.HostAnnouncement)
 	}, nil
 }
 
-func (e *Explorer) addHostScans(hosts chan chain.HostAnnouncement) {
+func (e *Explorer) scanV2Host(locator geoip.Locator, host Host) (HostScan, error) {
+	ctx, cancel := context.WithTimeout(e.ctx, e.scanCfg.Timeout)
+	defer cancel()
+
+	addr, ok := host.V2SiamuxAddr()
+	if !ok {
+		return HostScan{}, fmt.Errorf("host has no v2 siamux address")
+	}
+
+	transport, err := crhpv4.DialSiaMux(ctx, addr, host.PublicKey)
+	if err != nil {
+		return HostScan{}, fmt.Errorf("failed to dial host: %w", err)
+	}
+	defer transport.Close()
+
+	settings, err := crhpv4.RPCSettings(ctx, transport)
+	if err != nil {
+		return HostScan{}, fmt.Errorf("failed to get host settings: %w", err)
+	}
+
+	hostIP, _, err := net.SplitHostPort(addr)
+	if err != nil {
+		return HostScan{}, fmt.Errorf("scanHost: failed to parse net address: %w", err)
+	}
+
+	resolved, err := net.ResolveIPAddr("ip", hostIP)
+	if err != nil {
+		return HostScan{}, fmt.Errorf("scanHost: failed to resolve host address: %w", err)
+	}
+
+	countryCode, err := locator.CountryCode(resolved)
+	if err != nil {
+		e.log.Debug("Failed to resolve IP geolocation, not setting country code", zap.String("addr", host.NetAddress))
+		countryCode = ""
+	}
+
+	return HostScan{
+		PublicKey:   host.PublicKey,
+		CountryCode: countryCode,
+		Success:     true,
+		Timestamp:   types.CurrentTimestamp(),
+
+		RHPV4Settings: settings,
+	}, nil
+}
+
+func (e *Explorer) addHostScans(hosts chan Host) {
 	// use default included ip2location database
 	locator, err := geoip.NewIP2LocationLocator("")
 	if err != nil {
@@ -129,18 +175,32 @@ func (e *Explorer) addHostScans(hosts chan chain.HostAnnouncement) {
 				break
 			}
 
-			scan, err := e.scanHost(locator, host)
+			var scan HostScan
+			var addr string
+			var ok bool
+			var err error
+
+			if host.IsV2() {
+				addr, ok = host.V2SiamuxAddr()
+				if !ok {
+					e.log.Debug("Host did not have any v2 siamux net addresses in its announcement, unable to scan", zap.Stringer("pk", host.PublicKey))
+					continue
+				}
+				scan, err = e.scanV2Host(locator, host)
+			} else {
+				scan, err = e.scanV1Host(locator, host)
+			}
 			if err != nil {
 				scans = append(scans, HostScan{
 					PublicKey: host.PublicKey,
 					Success:   false,
 					Timestamp: types.CurrentTimestamp(),
 				})
-				e.log.Debug("Scanning host failed", zap.String("addr", host.NetAddress), zap.Stringer("pk", host.PublicKey), zap.Error(err))
+				e.log.Debug("Scanning host failed", zap.String("addr", addr), zap.Stringer("pk", host.PublicKey), zap.Error(err))
 				continue
 			}
 
-			e.log.Debug("Scanning host succeeded", zap.String("addr", host.NetAddress), zap.Stringer("pk", host.PublicKey))
+			e.log.Debug("Scanning host succeeded", zap.String("addr", addr), zap.Stringer("pk", host.PublicKey))
 			scans = append(scans, scan)
 		}
 
@@ -172,7 +232,7 @@ func (e *Explorer) isClosed() bool {
 	}
 }
 
-func (e *Explorer) fetchHosts(hosts chan chain.HostAnnouncement) {
+func (e *Explorer) fetchHosts(hosts chan Host) {
 	var exhausted bool
 	offset := 0
 
@@ -210,7 +270,7 @@ func (e *Explorer) scanHosts() {
 
 	for !e.isClosed() {
 		// fetch hosts
-		hosts := make(chan chain.HostAnnouncement, scanBatchSize)
+		hosts := make(chan Host, scanBatchSize)
 		e.wg.Add(1)
 		go func() {
 			defer e.wg.Done()

explorer/types.go

@@ -8,8 +8,10 @@ import (
 	"go.sia.tech/core/consensus"
 	rhpv2 "go.sia.tech/core/rhp/v2"
 	rhpv3 "go.sia.tech/core/rhp/v3"
+	rhpv4 "go.sia.tech/core/rhp/v4"
 	"go.sia.tech/core/types"
 	"go.sia.tech/coreutils/chain"
+	crhpv4 "go.sia.tech/coreutils/rhp/v4"
 )
 
 // A Source represents where a siacoin output came from.
@@ -287,6 +289,8 @@ type HostScan struct {
 
 	Settings   rhpv2.HostSettings   `json:"settings"`
 	PriceTable rhpv3.HostPriceTable `json:"priceTable"`
+
+	RHPV4Settings rhpv4.HostSettings `json:"rhpV4Settings"`
 }
 
 // Host represents a host and the information gathered from scanning it.
@@ -307,6 +311,25 @@ type Host struct {
 
 	Settings   rhpv2.HostSettings   `json:"settings"`
 	PriceTable rhpv3.HostPriceTable `json:"priceTable"`
+
+	RHPV4Settings rhpv4.HostSettings `json:"rhpV4Settings"`
+}
+
+// V2SiamuxAddr returns the `Address` of the first TCP siamux `NetAddress` it
+// finds in the host's list of net addresses.  The protocol for this address is
+// ProtocolTCPSiaMux.
+func (h Host) V2SiamuxAddr() (string, bool) {
+	for _, netAddr := range h.V2NetAddresses {
+		if netAddr.Protocol == crhpv4.ProtocolTCPSiaMux {
+			return netAddr.Address, true
+		}
+	}
+	return "", false
+}
+
+// IsV2 returns whether a host supports V2 or not.
+func (h Host) IsV2() bool {
+	return len(h.V2NetAddresses) > 0
 }
 
 // HostMetrics represents averages of scanned information from hosts.

persist/sqlite/addresses.go

@@ -89,7 +89,7 @@ func (st *Store) Hosts(pks []types.PublicKey) (result []explorer.Host, err error
 			encoded = append(encoded, encode(pk))
 		}
 
-		rows, err := tx.Query(`SELECT public_key,net_address,country_code,known_since,last_scan,last_scan_successful,last_announcement,total_scans,successful_interactions,failed_interactions,settings_accepting_contracts,settings_max_download_batch_size,settings_max_duration,settings_max_revise_batch_size,settings_net_address,settings_remaining_storage,settings_sector_size,settings_total_storage,settings_address,settings_window_size,settings_collateral,settings_max_collateral,settings_base_rpc_price,settings_contract_price,settings_download_bandwidth_price,settings_sector_access_price,settings_storage_price,settings_upload_bandwidth_price,settings_ephemeral_account_expiry,settings_max_ephemeral_account_balance,settings_revision_number,settings_version,settings_release,settings_sia_mux_port,price_table_uid,price_table_validity,price_table_host_block_height,price_table_update_price_table_cost,price_table_account_balance_cost,price_table_fund_account_cost,price_table_latest_revision_cost,price_table_subscription_memory_cost,price_table_subscription_notification_cost,price_table_init_base_cost,price_table_memory_time_cost,price_table_download_bandwidth_cost,price_table_upload_bandwidth_cost,price_table_drop_sectors_base_cost,price_table_drop_sectors_unit_cost,price_table_has_sector_base_cost,price_table_read_base_cost,price_table_read_length_cost,price_table_renew_contract_cost,price_table_revision_base_cost,price_table_swap_sector_base_cost,price_table_write_base_cost,price_table_write_length_cost,price_table_write_store_cost,price_table_txn_fee_min_recommended,price_table_txn_fee_max_recommended,price_table_contract_price,price_table_collateral_cost,price_table_max_collateral,price_table_max_duration,price_table_window_size,price_table_registry_entries_left,price_table_registry_entries_total FROM host_info WHERE public_key IN (`+queryPlaceHolders(len(pks))+`)`, encoded...)
+		rows, err := tx.Query(`SELECT public_key,net_address,country_code,known_since,last_scan,last_scan_successful,last_announcement,total_scans,successful_interactions,failed_interactions,settings_accepting_contracts,settings_max_download_batch_size,settings_max_duration,settings_max_revise_batch_size,settings_net_address,settings_remaining_storage,settings_sector_size,settings_total_storage,settings_address,settings_window_size,settings_collateral,settings_max_collateral,settings_base_rpc_price,settings_contract_price,settings_download_bandwidth_price,settings_sector_access_price,settings_storage_price,settings_upload_bandwidth_price,settings_ephemeral_account_expiry,settings_max_ephemeral_account_balance,settings_revision_number,settings_version,settings_release,settings_sia_mux_port,price_table_uid,price_table_validity,price_table_host_block_height,price_table_update_price_table_cost,price_table_account_balance_cost,price_table_fund_account_cost,price_table_latest_revision_cost,price_table_subscription_memory_cost,price_table_subscription_notification_cost,price_table_init_base_cost,price_table_memory_time_cost,price_table_download_bandwidth_cost,price_table_upload_bandwidth_cost,price_table_drop_sectors_base_cost,price_table_drop_sectors_unit_cost,price_table_has_sector_base_cost,price_table_read_base_cost,price_table_read_length_cost,price_table_renew_contract_cost,price_table_revision_base_cost,price_table_swap_sector_base_cost,price_table_write_base_cost,price_table_write_length_cost,price_table_write_store_cost,price_table_txn_fee_min_recommended,price_table_txn_fee_max_recommended,price_table_contract_price,price_table_collateral_cost,price_table_max_collateral,price_table_max_duration,price_table_window_size,price_table_registry_entries_left,price_table_registry_entries_total,rhp4_settings_protocol_version,rhp4_settings_release,rhp4_settings_wallet_address,rhp4_settings_accepting_contracts,rhp4_settings_max_collateral,rhp4_settings_max_collateral_duration,rhp4_settings_max_sector_duration,rhp4_settings_max_sector_batch_size,rhp4_settings_remaining_storage,rhp4_settings_total_storage,rhp4_prices_contract_price,rhp4_prices_collateral_price,rhp4_prices_storage_price,rhp4_prices_ingress_price,rhp4_prices_egress_price,rhp4_prices_free_sector_price,rhp4_prices_tip_height,rhp4_prices_valid_until,rhp4_prices_signature FROM host_info WHERE public_key IN (`+queryPlaceHolders(len(pks))+`)`, encoded...)
 		if err != nil {
 			return err
 		}
@@ -104,10 +104,14 @@ func (st *Store) Hosts(pks []types.PublicKey) (result []explorer.Host, err error
 		for rows.Next() {
 			if err := func() error {
 				var host explorer.Host
+				var protocolVersion []uint8
+
 				s, p := &host.Settings, &host.PriceTable
-				if err := rows.Scan(decode(&host.PublicKey), &host.NetAddress, &host.CountryCode, decode(&host.KnownSince), decode(&host.LastScan), &host.LastScanSuccessful, decode(&host.LastAnnouncement), &host.TotalScans, &host.SuccessfulInteractions, &host.FailedInteractions, &s.AcceptingContracts, decode(&s.MaxDownloadBatchSize), decode(&s.MaxDuration), decode(&s.MaxReviseBatchSize), &s.NetAddress, decode(&s.RemainingStorage), decode(&s.SectorSize), decode(&s.TotalStorage), decode(&s.Address), decode(&s.WindowSize), decode(&s.Collateral), decode(&s.MaxCollateral), decode(&s.BaseRPCPrice), decode(&s.ContractPrice), decode(&s.DownloadBandwidthPrice), decode(&s.SectorAccessPrice), decode(&s.StoragePrice), decode(&s.UploadBandwidthPrice), &s.EphemeralAccountExpiry, decode(&s.MaxEphemeralAccountBalance), decode(&s.RevisionNumber), &s.Version, &s.Release, &s.SiaMuxPort, decode(&p.UID), &p.Validity, decode(&p.HostBlockHeight), decode(&p.UpdatePriceTableCost), decode(&p.AccountBalanceCost), decode(&p.FundAccountCost), decode(&p.LatestRevisionCost), decode(&p.SubscriptionMemoryCost), decode(&p.SubscriptionNotificationCost), decode(&p.InitBaseCost), decode(&p.MemoryTimeCost), decode(&p.DownloadBandwidthCost), decode(&p.UploadBandwidthCost), decode(&p.DropSectorsBaseCost), decode(&p.DropSectorsUnitCost), decode(&p.HasSectorBaseCost), decode(&p.ReadBaseCost), decode(&p.ReadLengthCost), decode(&p.RenewContractCost), decode(&p.RevisionBaseCost), decode(&p.SwapSectorBaseCost), decode(&p.WriteBaseCost), decode(&p.WriteLengthCost), decode(&p.WriteStoreCost), decode(&p.TxnFeeMinRecommended), decode(&p.TxnFeeMaxRecommended), decode(&p.ContractPrice), decode(&p.CollateralCost), decode(&p.MaxCollateral), decode(&p.MaxDuration), decode(&p.WindowSize), decode(&p.RegistryEntriesLeft), decode(&p.RegistryEntriesTotal)); err != nil {
+				sV4, pV4 := &host.RHPV4Settings, &host.RHPV4Settings.Prices
+				if err := rows.Scan(decode(&host.PublicKey), &host.NetAddress, &host.CountryCode, decode(&host.KnownSince), decode(&host.LastScan), &host.LastScanSuccessful, decode(&host.LastAnnouncement), &host.TotalScans, &host.SuccessfulInteractions, &host.FailedInteractions, &s.AcceptingContracts, decode(&s.MaxDownloadBatchSize), decode(&s.MaxDuration), decode(&s.MaxReviseBatchSize), &s.NetAddress, decode(&s.RemainingStorage), decode(&s.SectorSize), decode(&s.TotalStorage), decode(&s.Address), decode(&s.WindowSize), decode(&s.Collateral), decode(&s.MaxCollateral), decode(&s.BaseRPCPrice), decode(&s.ContractPrice), decode(&s.DownloadBandwidthPrice), decode(&s.SectorAccessPrice), decode(&s.StoragePrice), decode(&s.UploadBandwidthPrice), &s.EphemeralAccountExpiry, decode(&s.MaxEphemeralAccountBalance), decode(&s.RevisionNumber), &s.Version, &s.Release, &s.SiaMuxPort, decode(&p.UID), &p.Validity, decode(&p.HostBlockHeight), decode(&p.UpdatePriceTableCost), decode(&p.AccountBalanceCost), decode(&p.FundAccountCost), decode(&p.LatestRevisionCost), decode(&p.SubscriptionMemoryCost), decode(&p.SubscriptionNotificationCost), decode(&p.InitBaseCost), decode(&p.MemoryTimeCost), decode(&p.DownloadBandwidthCost), decode(&p.UploadBandwidthCost), decode(&p.DropSectorsBaseCost), decode(&p.DropSectorsUnitCost), decode(&p.HasSectorBaseCost), decode(&p.ReadBaseCost), decode(&p.ReadLengthCost), decode(&p.RenewContractCost), decode(&p.RevisionBaseCost), decode(&p.SwapSectorBaseCost), decode(&p.WriteBaseCost), decode(&p.WriteLengthCost), decode(&p.WriteStoreCost), decode(&p.TxnFeeMinRecommended), decode(&p.TxnFeeMaxRecommended), decode(&p.ContractPrice), decode(&p.CollateralCost), decode(&p.MaxCollateral), decode(&p.MaxDuration), decode(&p.WindowSize), decode(&p.RegistryEntriesLeft), decode(&p.RegistryEntriesTotal), &protocolVersion, &sV4.Release, decode(&sV4.WalletAddress), &sV4.AcceptingContracts, decode(&sV4.MaxCollateral), decode(&sV4.MaxContractDuration), decode(&sV4.MaxSectorDuration), decode(&sV4.MaxSectorBatchSize), decode(&sV4.RemainingStorage), decode(&sV4.TotalStorage), decode(&pV4.ContractPrice), decode(&pV4.Collateral), decode(&pV4.StoragePrice), decode(&pV4.IngressPrice), decode(&pV4.EgressPrice), decode(&pV4.FreeSectorPrice), decode(&pV4.TipHeight), decode(&pV4.ValidUntil), decode(&pV4.Signature)); err != nil {
 					return err
 				}
+				sV4.ProtocolVersion = [3]uint8(protocolVersion)
 
 				v2AddrRows, err := v2AddrStmt.Query(encode(host.PublicKey))
 				if err != nil {
@@ -134,19 +138,47 @@ func (st *Store) Hosts(pks []types.PublicKey) (result []explorer.Host, err error
 }
 
 // HostsForScanning returns hosts ordered by the transaction they were created in.
-func (s *Store) HostsForScanning(maxLastScan, minLastAnnouncement time.Time, offset, limit uint64) (result []chain.HostAnnouncement, err error) {
+// Note that only the PublicKey, NetAddress, and V2NetAddresses fields are
+// populated.
+func (s *Store) HostsForScanning(maxLastScan, minLastAnnouncement time.Time, offset, limit uint64) (result []explorer.Host, err error) {
 	err = s.transaction(func(tx *txn) error {
 		rows, err := tx.Query(`SELECT public_key, net_address FROM host_info WHERE last_scan <= ? AND last_announcement >= ? ORDER BY last_scan ASC LIMIT ? OFFSET ?`, encode(maxLastScan), encode(minLastAnnouncement), limit, offset)
 		if err != nil {
 			return err
 		}
 		defer rows.Close()
 
+		v2AddrStmt, err := tx.Prepare(`SELECT protocol,address FROM host_info_v2_netaddresses WHERE public_key = ? ORDER BY netaddress_order`)
+		if err != nil {
+			return err
+		}
+		defer v2AddrStmt.Close()
+
 		for rows.Next() {
-			var host chain.HostAnnouncement
+			var host explorer.Host
 			if err := rows.Scan(decode(&host.PublicKey), &host.NetAddress); err != nil {
 				return err
 			}
+
+			err := func() error {
+				v2AddrRows, err := v2AddrStmt.Query(encode(host.PublicKey))
+				if err != nil {
+					return err
+				}
+				defer v2AddrRows.Close()
+				for v2AddrRows.Next() {
+					var netAddr chain.NetAddress
+					if err := v2AddrRows.Scan(&netAddr.Protocol, &netAddr.Address); err != nil {
+						return err
+					}
+					host.V2NetAddresses = append(host.V2NetAddresses, netAddr)
+				}
+				return nil
+			}()
+			if err != nil {
+				return err
+			}
+
 			result = append(result, host)
 		}
 		return nil