diff --git a/harmony/harmonydb/downgrade/20260821-drop-ipni-ad-fetches.sql b/harmony/harmonydb/downgrade/20260821-drop-ipni-ad-fetches.sql new file mode 100644 index 000000000..56f843919 --- /dev/null +++ b/harmony/harmonydb/downgrade/20260821-drop-ipni-ad-fetches.sql @@ -0,0 +1,6 @@ +CREATE TABLE IF NOT EXISTS ipni_ad_fetches ( + ad_cid TEXT NOT NULL, + fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS ipni_ad_fetches_ad_cid_time ON ipni_ad_fetches(ad_cid, fetched_at DESC); diff --git a/harmony/harmonydb/sql/20260821-drop-ipni-ad-fetches.sql b/harmony/harmonydb/sql/20260821-drop-ipni-ad-fetches.sql new file mode 100644 index 000000000..41fa007c4 --- /dev/null +++ b/harmony/harmonydb/sql/20260821-drop-ipni-ad-fetches.sql @@ -0,0 +1,7 @@ +-- ipni_ad_fetches recorded raw ad-body fetches, which only proves an indexer +-- pulled the ad, not that it finished processing it. The piece-status +-- endpoint now returns the ad's CID directly so callers can check sync status +-- against the indexer themselves, and "advertised" is tracked in memory by +-- the ipni-provider process (see market/ipni/ipni-provider), so nothing here +-- needs replacing. +DROP TABLE IF EXISTS ipni_ad_fetches; diff --git a/market/ipni/ipni-provider/ipni-provider.go b/market/ipni/ipni-provider/ipni-provider.go index 32e18a0df..5f2a26f3e 100644 --- a/market/ipni/ipni-provider/ipni-provider.go +++ b/market/ipni/ipni-provider/ipni-provider.go @@ -51,7 +51,7 @@ const IPNIRoutePath = "/ipni-provider/" // IPNIPath is a constant that represents the path for IPNI API requests. const IPNIPath = "/ipni/v1/ad/" -const PublishInterval = 5 * time.Second +const PublishInterval = 1 * time.Second const ( PDPv0ProviderType = "PDP_v0" @@ -460,8 +460,6 @@ func (p *Provider) handleGet(w http.ResponseWriter, r *http.Request) { if err != nil { log.Errorw("failed to write HTTP response", "err", err) } - // Log advertisement fetch for indexing status tracking - go p.logPDPFetch(providerID, b.String()) return case ipnisync.CidSchemaEntryChunk: content = "entry" @@ -522,26 +520,10 @@ func (p *Provider) handleGet(w http.ResponseWriter, r *http.Request) { if err != nil { log.Errorw("failed to write HTTP response", "err", err) } - // Log advertisement fetch for indexing status tracking - go p.logPDPFetch(providerID, b.String()) return } } -func (p *Provider) logPDPFetch(peer, b string) { - p.mu.RLock() - info, ok := p.providerInfos[peer] - p.mu.RUnlock() - if !ok || info.SPID > 0 { - return - } - logCtx := context.Background() - _, err := p.db.Exec(logCtx, `INSERT INTO ipni_ad_fetches (ad_cid, fetched_at) VALUES ($1, NOW())`, b) - if err != nil { - log.Warnw("failed to log ad fetch", "ad_cid", b, "err", err) - } -} - // Routes sets up the routes for the IPNI provider. // It registers a handler function for the GET request at the IPNIRoutePath. // The handler function is provided by the Provider struct. @@ -589,22 +571,6 @@ func (p *Provider) startPublishing(ctx context.Context) { p.announceURLs = urls } - // Populated latest head cid from the ipni_head table - p.mu.RLock() - peers := make([]string, 0, len(p.providerInfos)) - for pr := range p.providerInfos { - peers = append(peers, pr) - } - p.mu.RUnlock() - for _, provider := range peers { - c, err := p.getHeadCID(ctx, provider) - if err != nil { - log.Errorw("failed to get head CID", "provider", provider, "error", err) - continue - } - p.latest[provider] = c - } - ticker := time.NewTicker(PublishInterval) for { diff --git a/pdp/README.md b/pdp/README.md index c43b29a8d..d63e1f86b 100644 --- a/pdp/README.md +++ b/pdp/README.md @@ -162,9 +162,12 @@ All endpoints are rooted at `/pdp`. "pieceCid": "", "status": "", "indexed": , + "indexedAt": "", + "adCreated": , + "adCreatedAt": "", + "adCid": "", "advertised": , - "retrieved": , - "retrievedAt": "" + "advertisedAt": "" } ``` @@ -174,12 +177,14 @@ All endpoints are rooted at `/pdp`. - `"pending"` – Not yet indexed. - `"indexing"` – CAR indexing task is in progress. - `"creating_ad"` – IPNI advertisement is being created. - - `"announced"` – Advertisement published to IPNI network. - - `"retrieved"` – Piece has been retrieved by a client. + - `"announced"` – The advertisement row exists. Not a guarantee it's been broadcast (see `advertised`) or indexed (see `adCid`). - `indexed`: Whether the piece has been indexed and is ready for IPNI. - - `advertised`: Whether an IPNI advertisement has been published. - - `retrieved`: Whether the piece has been retrieved by a client. - - `retrievedAt`: Timestamp of last retrieval (omitted if never retrieved). + - `indexedAt`: Timestamp CAR indexing completed (omitted if not yet indexed). + - `adCreated`: Whether an IPNI advertisement has been created for this piece. + - `adCreatedAt`: Timestamp the advertisement was created (omitted if not yet created). + - `adCid`: This piece's advertisement CID, once created. Curio doesn't check whether an indexer finished processing it - callers can, e.g. `GET https://cid.contact/sync/status/ad/{adCid}`. + - `advertised`: Whether the provider has sent an HTTP announce covering this ad. + - `advertisedAt`: Approximate, not a fixed record: it's the last known successful announce time for the *provider*, not this ad specifically, so it can drift forward on later calls once the provider announces newer ads, and is lost on a Curio restart (the provider re-announces its current head once on startup to recover it). #### Errors diff --git a/pdp/handlers.go b/pdp/handlers.go index ce451b356..244fcba3e 100644 --- a/pdp/handlers.go +++ b/pdp/handlers.go @@ -270,18 +270,16 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request // Query status from database var results []struct { - PieceCID string `db:"piece_cid"` - PieceRawSize uint64 `db:"piece_raw_size"` - CreatedAt time.Time `db:"created_at"` - Indexed bool `db:"indexed"` - IndexedAt sql.NullTime `db:"indexed_at"` - AdvertisementCreated bool `db:"advertisement_created"` - AdvertisementCreatedAt sql.NullTime `db:"advertisement_created_at"` - AdCID sql.NullString `db:"ad_cid"` - AdvertisementRetrieved bool `db:"advertisement_retrieved"` - AdvertisementRetrievedAt sql.NullTime `db:"advertisement_retrieved_at"` - Status string `db:"status"` - Provider sql.NullString `db:"provider"` + PieceCID string `db:"piece_cid"` + PieceRawSize uint64 `db:"piece_raw_size"` + CreatedAt time.Time `db:"created_at"` + Indexed bool `db:"indexed"` + IndexedAt sql.NullTime `db:"indexed_at"` + AdvertisementCreated bool `db:"advertisement_created"` + AdvertisementCreatedAt sql.NullTime `db:"advertisement_created_at"` + AdCID sql.NullString `db:"ad_cid"` + Status string `db:"status"` + Provider sql.NullString `db:"provider"` } err = p.db.Select(ctx, &results, ` @@ -299,19 +297,16 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request pr.advertisement_created_at as advertisement_created_at, ia.ad_cid, - -- Advertisement Fetch status - ia.fetched_at IS NOT NULL as advertisement_retrieved, - ia.fetched_at as advertisement_retrieved_at, - - -- Determine overall status + -- Determine overall status. "announced" only means the ad row + -- exists, not that an indexer finished processing it; callers + -- check that themselves via adCid. CASE - WHEN ia.fetched_at IS NOT NULL THEN 'retrieved' WHEN ia.ad_cid IS NOT NULL THEN 'announced' WHEN pr.ipni_task_id IS NOT NULL THEN 'creating_ad' WHEN pr.indexing_task_id IS NOT NULL THEN 'indexing' ELSE 'pending' END as status, - + ia.provider FROM pdp_piecerefs pr @@ -320,8 +315,7 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request LEFT JOIN LATERAL ( SELECT MIN(i.ad_cid) as ad_cid, - MIN(i.provider) as provider, - MIN((SELECT MIN(af.fetched_at) FROM ipni_ad_fetches af WHERE af.ad_cid = i.ad_cid)) as fetched_at + MIN(i.provider) as provider FROM ipni i WHERE i.piece_cid = pr.piece_cid AND i.provider = (SELECT peer_id FROM ipni_peerid WHERE sp_id = $3) @@ -357,16 +351,14 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request IndexedAt *time.Time `json:"indexedAt,omitempty"` AdCreated bool `json:"adCreated"` AdCreatedAt *time.Time `json:"adCreatedAt,omitempty"` + AdCid *string `json:"adCid,omitempty"` Advertised bool `json:"advertised"` AdvertisedAt *time.Time `json:"advertisedAt,omitempty"` - Retrieved bool `json:"retrieved"` - RetrievedAt *time.Time `json:"retrievedAt,omitempty"` }{ PieceCID: pieceInfo.CidV2.String(), Status: result.Status, Indexed: result.Indexed, AdCreated: result.AdvertisementCreated, - Retrieved: result.AdvertisementRetrieved, } if !result.IndexedAt.Valid { @@ -381,72 +373,26 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request response.AdCreatedAt = &result.AdvertisementCreatedAt.Time } - if !result.AdvertisementRetrievedAt.Valid { - response.RetrievedAt = nil - } else { - response.RetrievedAt = &result.AdvertisementRetrievedAt.Time - } - - // Advertised and AdvertisedAt are derived from three signals, in order: - // 1. A recorded fetch of this ad in ipni_ad_fetches is the strongest per-ad - // signal. If an indexer fetched the ad, it must have been advertised - // already. Since we do not store the actual first publish time for each - // ad, AdvertisedAt is estimated as the earlier of - // advertisement_created_at + PublishInterval and the first fetch time. - // This keeps AdvertisedAt from appearing after RetrievedAt. - // 2. The in-process IPNI provider exposes LastPublishTime per provider, not - // per ad. It is useful only when there is no fetch record and we know - // when this ad was created. A provider publish after this ad was created - // means the ad should have been included in the announced head; an older - // provider publish means it was not, and we do not fall through to the - // timing heuristic. - // 3. Without either signal, fall back to the old timing heuristic: after - // PublishInterval has elapsed from ad creation, assume the ad was - // announced. - if result.AdvertisementRetrieved { - response.Advertised = true - if result.AdvertisementRetrievedAt.Valid { - advertisedAt := result.AdvertisementRetrievedAt.Time - if result.AdvertisementCreatedAt.Valid { - createdAtEstimate := result.AdvertisementCreatedAt.Time.Add(ipni_provider.PublishInterval) - if createdAtEstimate.Before(advertisedAt) { - advertisedAt = createdAtEstimate - } - } - response.AdvertisedAt = &advertisedAt - } else { - response.AdvertisedAt = nil - } + if result.AdCID.Valid { + response.AdCid = &result.AdCID.String } - advertisedFromProvider := false - if !response.Advertised && result.AdvertisementCreatedAt.Valid && p.ipp != nil && result.Provider.Valid { - publishedAt := p.ipp.LastPublishTime(result.Provider.String) + // LastPublishTime is per-provider, not per-ad, so it only confirms this ad + // was advertised once it postdates the ad's creation. Falls back to the + // old timing heuristic if that signal isn't available. + if result.AdvertisementCreatedAt.Valid { + var publishedAt *time.Time + if p.ipp != nil && result.Provider.Valid { + publishedAt = p.ipp.LastPublishTime(result.Provider.String) + } if publishedAt != nil { - advertisedFromProvider = true if publishedAt.After(result.AdvertisementCreatedAt.Time) { response.Advertised = true response.AdvertisedAt = new(*publishedAt) - if publishedAt.After(time.Now().Add(ipni_provider.PublishInterval)) { - response.AdvertisedAt = new(result.AdvertisementCreatedAt.Time.Add(ipni_provider.PublishInterval)) - } - } else { - response.Advertised = false - response.AdvertisedAt = nil - } - } - } - - if !advertisedFromProvider && !response.Advertised { - if result.AdvertisementCreated && result.AdvertisementCreatedAt.Valid { - if time.Since(result.AdvertisementCreatedAt.Time) > ipni_provider.PublishInterval { - // More than 5 seconds since advertisement was created, assume it's published - response.Advertised = true - response.AdvertisedAt = new(result.AdvertisementCreatedAt.Time.Add(ipni_provider.PublishInterval)) - } else { - response.Advertised = false - response.AdvertisedAt = nil } + } else if time.Since(result.AdvertisementCreatedAt.Time) > ipni_provider.PublishInterval { + response.Advertised = true + response.AdvertisedAt = new(result.AdvertisementCreatedAt.Time.Add(ipni_provider.PublishInterval)) } }