worker: update cache when hosts reannounce

api/events.go

@@ -14,6 +14,7 @@ const (
 	ModuleConsensus   = "consensus"
 	ModuleContract    = "contract"
 	ModuleContractSet = "contract_set"
+	ModuleHost        = "host"
 	ModuleSetting     = "setting"
 
 	EventUpdate  = "update"
@@ -44,6 +45,12 @@ type (
 		Timestamp time.Time        `json:"timestamp"`
 	}
 
+	EventHostUpdate struct {
+		HostKey   types.PublicKey `json:"hostKey"`
+		NetAddr   string          `json:"netAddr"`
+		Timestamp time.Time       `json:"timestamp"`
+	}
+
 	EventContractSetUpdate struct {
 		Name        string                 `json:"name"`
 		ContractIDs []types.FileContractID `json:"contractIDs"`
@@ -99,6 +106,15 @@ var (
 		}
 	}
 
+	WebhookHostUpdate = func(url string, headers map[string]string) webhooks.Webhook {
+		return webhooks.Webhook{
+			Event:   EventUpdate,
+			Headers: headers,
+			Module:  ModuleHost,
+			URL:     url,
+		}
+	}
+
 	WebhookSettingUpdate = func(url string, headers map[string]string) webhooks.Webhook {
 		return webhooks.Webhook{
 			Event:   EventUpdate,
@@ -155,6 +171,14 @@ func ParseEventWebhook(event webhooks.Event) (interface{}, error) {
 			}
 			return e, nil
 		}
+	case ModuleHost:
+		if event.Event == EventUpdate {
+			var e EventHostUpdate
+			if err := json.Unmarshal(bytes, &e); err != nil {
+				return nil, err
+			}
+			return e, nil
+		}
 	case ModuleSetting:
 		switch event.Event {
 		case EventUpdate:

internal/chain/chainsubscriber.go

@@ -170,6 +170,17 @@ func (s *ChainSubscriber) applyChainUpdate(tx ChainUpdateTx, cau chain.ApplyUpda
 		for hk, ha := range hus {
 			if err := tx.UpdateHost(hk, ha, cau.State.Index.Height, b.ID(), b.Timestamp); err != nil {
 				return fmt.Errorf("failed to update host: %w", err)
+			} else if IsSynced(b) {
+				// broadcast host update
+				s.webhooksMgr.BroadcastAction(s.shutdownCtx, webhooks.Event{
+					Module: api.ModuleHost,
+					Event:  api.EventUpdate,
+					Payload: api.EventHostUpdate{
+						HostKey:   hk,
+						NetAddr:   ha.NetAddress,
+						Timestamp: time.Now().UTC(),
+					},
+				})
 			}
 		}
 	}

internal/test/e2e/events_test.go

@@ -25,6 +25,7 @@ func TestEvents(t *testing.T) {
 		api.WebhookContractArchive,
 		api.WebhookContractRenew,
 		api.WebhookContractSetUpdate,
+		api.WebhookHostUpdate,
 		api.WebhookSettingDelete,
 		api.WebhookSettingUpdate,
 	}
@@ -49,6 +50,15 @@ func TestEvents(t *testing.T) {
 			return nil
 		}
 
+		// ignore host updates with net address diff. from the update we assert to receive
+		if event.Module == api.ModuleHost && event.Event == api.EventUpdate {
+			if parsed, err := api.ParseEventWebhook(event); err != nil {
+				t.Fatal(err)
+			} else if parsed.(api.EventHostUpdate).NetAddr != "127.0.0.1:0" {
+				return nil
+			}
+		}
+
 		// check if the event is expected
 		if !isKnownEvent(event) {
 			return fmt.Errorf("unexpected event %+v", event)
@@ -120,6 +130,12 @@ func TestEvents(t *testing.T) {
 	// delete setting
 	tt.OK(b.DeleteSetting(context.Background(), api.SettingRedundancy))
 
+	// update host setting
+	h := cluster.hosts[0]
+	settings := h.settings.Settings()
+	settings.NetAddress = "127.0.0.1:0"
+	tt.OK(h.UpdateSettings(settings))
+
 	// wait until we received the events
 	tt.Retry(10, time.Second, func() error {
 		mu.Lock()
@@ -152,6 +168,10 @@ func TestEvents(t *testing.T) {
 			if e.TransactionFee.IsZero() || e.BlockHeight == 0 || e.Timestamp.IsZero() || !e.Synced {
 				t.Fatalf("unexpected event %+v", e)
 			}
+		case api.EventHostUpdate:
+			if e.HostKey != h.PublicKey() || e.NetAddr != "127.0.0.1:0" || e.Timestamp.IsZero() {
+				t.Fatalf("unexpected event %+v", e)
+			}
 		case api.EventSettingUpdate:
 			if e.Key != api.SettingGouging || e.Timestamp.IsZero() {
 				t.Fatalf("unexpected event %+v", e)

internal/worker/cache.go

@@ -174,6 +174,9 @@ func (c *cache) HandleEvent(event webhooks.Event) (err error) {
 	case api.EventContractRenew:
 		log = log.With("fcid", e.Renewal.ID, "renewedFrom", e.Renewal.RenewedFrom, "ts", e.Timestamp)
 		c.handleContractRenew(e)
+	case api.EventHostUpdate:
+		log = log.With("hk", e.HostKey, "ts", e.Timestamp)
+		c.handleHostUpdate(e)
 	case api.EventSettingUpdate:
 		log = log.With("key", e.Key, "ts", e.Timestamp)
 		err = c.handleSettingUpdate(e)
@@ -204,6 +207,7 @@ func (c *cache) Initialize(ctx context.Context, workerAPI string, webhookOpts ..
 		api.WebhookConsensusUpdate(eventsURL, headers),
 		api.WebhookContractArchive(eventsURL, headers),
 		api.WebhookContractRenew(eventsURL, headers),
+		api.WebhookHostUpdate(eventsURL, headers),
 		api.WebhookSettingUpdate(eventsURL, headers),
 	} {
 		if err := c.b.RegisterWebhook(ctx, wh); err != nil {
@@ -273,6 +277,25 @@ func (c *cache) handleContractRenew(event api.EventContractRenew) {
 	c.cache.Set(cacheKeyDownloadContracts, contracts)
 }
 
+func (c *cache) handleHostUpdate(e api.EventHostUpdate) {
+	// return early if the cache doesn't have contracts
+	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)
+	if !found {
+		return
+	}
+	contracts := value.([]api.ContractMetadata)
+
+	// update the host's IP in the cache
+	for i, contract := range contracts {
+		if contract.HostKey == e.HostKey {
+			contracts[i].HostIP = e.NetAddr
+			break
+		}
+	}
+
+	c.cache.Set(cacheKeyDownloadContracts, contracts)
+}
+
 func (c *cache) handleSettingDelete(e api.EventSettingDelete) {
 	if e.Key == api.SettingGouging || e.Key == api.SettingRedundancy {
 		c.cache.Invalidate(cacheKeyGougingParams)

internal/worker/cache_test.go

@@ -149,6 +149,7 @@ func TestWorkerCache(t *testing.T) {
 		{Module: api.ModuleConsensus, Event: api.EventUpdate, Payload: nil},
 		{Module: api.ModuleContract, Event: api.EventArchive, Payload: nil},
 		{Module: api.ModuleContract, Event: api.EventRenew, Payload: nil},
+		{Module: api.ModuleHost, Event: api.EventUpdate, Payload: nil},
 		{Module: api.ModuleSetting, Event: api.EventUpdate, Payload: nil},
 		{Module: api.ModuleSetting, Event: api.EventDelete, Payload: nil},
 	} {