Sistemi distribuiti

Reading time: 5 min

The Graph è un protocollo implementato come sistema distribuito.

Le connessioni falliscono. Le richieste arrivano in ordine sparso. Computer diversi con orologi e stati non sincronizzati elaborano richieste correlate. I server si riavviano. Tra una richiesta e l'altra si verificano delle riorganizzazioni. Questi problemi sono inerenti a tutti i sistemi distribuiti, ma sono esacerbati nei sistemi che operano su scala globale.

Si consideri questo esempio di ciò che può accadere se un cliente interroga un Indexer per ottenere i dati più recenti durante una riorganizzazione.

1. L'Indexer inserisce il blocco 8
2. Richiesta servita al cliente per il blocco 8
3. L'Indexer inserisce il blocco 9
4. L'Indexer inserisce il blocco 10A
5. Richiesta servita al cliente per il blocco 10A
6. L'Indexer rileva la riorganizzazione a 10B e fa rollback di 10A
7. Richiesta servita al cliente per il blocco 9
8. L'Indexer inserisce il blocco 10B
9. L'Indexer inserisce il blocco 11
10. Richiesta servita al cliente per il blocco 11

Dal punto di vista dell'Indexer, le cose stanno procedendo in modo logico. Il tempo sta avanzando, anche se abbiamo dovuto fare un rollback di un blocco sconosciuto e riprodurre il blocco sotto consenso in avanti sopra di esso. Lungo il percorso, l'Indexer serve le richieste utilizzando lo stato più recente che conosce in quel momento.

Dal punto di vista del cliente, tuttavia, la situazione appare caotica. Il cliente osserva che le risposte sono state date per i blocchi 8, 10, 9 e 11 in questo ordine. Questo è il cosiddetto problema del "block wobble". Quando un cliente sperimenta il block wobble, i dati possono sembrare contraddirsi nel tempo. La situazione peggiora se si considera che gli Indexer non ingeriscono tutti i blocchi più recenti contemporaneamente e le richieste possono essere indirizzate a più Indexer.

È responsabilità del cliente e del server lavorare insieme per fornire dati coerenti all'utente. È necessario utilizzare approcci diversi a seconda della coerenza desiderata, poiché non esiste un unico programma giusto per ogni problema.

Ragionare sulle implicazioni dei sistemi distribuiti è difficile, ma la soluzione potrebbe non esserlo! Abbiamo creato delle API e dei modelli per aiutarvi a navigare in alcuni casi d'uso comuni. Gli esempi che seguono illustrano questi modelli, ma eludono i dettagli richiesti dal codice di produzione (come la gestione degli errori e la cancellazione) per non offuscare le idee principali.

The Graph fornisce l'API di blocco: { number_gte: $minBlock } che assicura che la risposta sia per un singolo blocco uguale o superiore a $minBlock. Se la richiesta è fatta a un'istanza di graph-node e il blocco min non è ancora sincronizzato, graph-node restituirà un errore. Se graph-node ha sincronizzato il blocco min, eseguirà la risposta per il blocco più recente. Se la richiesta viene fatta a un Edge&Node Gateway, il Gateway filtrerà tutti gli Indexer che non hanno ancora sincronizzato il blocco min e farà la richiesta per l'ultimo blocco che l'Indexer ha sincronizzato.

Possiamo usare numero_gte per assicurarci che il tempo non viaggi mai all'indietro durante il polling dei dati in un ciclo. Ecco un esempio:

/// Updates the protocol.paused variable to the latest
/// known value in a loop by fetching it using The Graph.
async function updateProtocolPaused() {
  // It's ok to start with minBlock at 0. The query will be served
  // using the latest block available. Setting minBlock to 0 is the
  // same as leaving out that argument.
  let minBlock = 0

  for (;;) {
    // Schedule a promise that will be ready once
    // the next Ethereum block will likely be available.
    const nextBlock = new Promise((f) => {
      setTimeout(f, 14000)
    })

    const query = `
        query GetProtocol($minBlock: Int!) {
            protocol(block: { number_gte: $minBlock }  id: "0") {
              paused
            }
            _meta {
                block {
                    number
                }
            }
        }`

    const variables = { minBlock }
    const response = await graphql(query, variables)
    minBlock = response._meta.block.number

    // TODO: Do something with the response data here instead of logging it.
    console.log(response.protocol.paused)

    // Sleep to wait for the next block
    await nextBlock
  }
}

Un altro caso d'uso è il recupero di un grande serie oppure, più in generale, il recupero di elementi correlati tra più richieste. A differenza del caso del polling (in cui la consistenza desiderata era quella di andare avanti nel tempo), la consistenza desiderata riguarda un singolo punto nel tempo.

Qui useremo l'argomento block: { hash: $blockHash } per fissare tutti i risultati allo stesso blocco.

/// Gets a list of domain names from a single block using pagination
async function getDomainNames() {
  // Set a cap on the maximum number of items to pull.
  let pages = 5
  const perPage = 1000

  // The first query will get the first page of results and also get the block
  // hash so that the remainder of the queries are consistent with the first.
  const listDomainsQuery = `
    query ListDomains($perPage: Int!) {
        domains(first: $perPage) {
            name
            id
        }
        _meta {
            block {
                hash
            }
        }
    }`

  let data = await graphql(listDomainsQuery, { perPage })
  let result = data.domains.map((d) => d.name)
  let blockHash = data._meta.block.hash

  let query
  // Continue fetching additional pages until either we run into the limit of
  // 5 pages total (specified above) or we know we have reached the last page
  // because the page has fewer entities than a full page.
  while (data.domains.length == perPage && --pages) {
    let lastID = data.domains[data.domains.length - 1].id
    query = `
        query ListDomains($perPage: Int!, $lastID: ID!, $blockHash: Bytes!) {
            domains(first: $perPage, where: { id_gt: $lastID }, block: { hash: $blockHash }) {
                name
                id
            }
        }`

    data = await graphql(query, { perPage, lastID, blockHash })

    // Accumulate domain names into the result
    for (domain of data.domains) {
      result.push(domain.name)
    }
  }
  return result
}

Si noti che, in caso di riorganizzazione, il cliente dovrà riprovare a partire dalla prima richiesta di aggiornamento dell'hash del blocco a un blocco non-uncle.