Couchbase Eventing Handling Errors and Retries w/ Examples

A Serviço de Eventing da Couchbase permite que você atue rapidamente em relação a mutações (ou alterações) em seus dados. Todas as ações no Adestramento Completo São realizadas através da execução de um lambda, um pequeno pedaço de lógica de negócio escrito em JavaScript.

Os casos de uso mais comuns incluem enriquecimento de dados, arquivamento de documentos e integração com serviços externos REST. Veja mais detalhes aqui.

No blog a seguir, descreveremos como os erros durante a execução do listener de eventos podem ser tratados. Usando um mecanismo de retorno, garantimos que a ação prevista seja realizada, mesmo que o listener de eventos falhe durante a execução.

Exemplo de aplicativo

Como exemplo, implementamos parte de um aplicativo de e-commerce que armazena os pedidos dos clientes em uma coleção do Couchbase. Assim que o status de uma ordem for alterado para Pago, Queremos enviar uma confirmação de encomenda ao cliente.

Para conseguir isso, integramos um listener de eventos do Couchbase com um serviço externo. serviço de e-mail. O listener de eventos do Couchbase irá captar quaisquer alterações no documento de ordem, verificar se a ordem foi paga e, em seguida, solicitar ao serviço de e-mail que acione a mensagem de confirmação.

O serviço de e-mail é um microserviço independente que fornece um endpoint REST. Utilizamos o suporte cURL integrado diretamente ao Couchbase Eventing para chamar o microserviço a partir do listener de eventos.

Fluxo conceitual: À medida que as ordens são atualizadas no Couchbase, eventos são ativados e captados por um listener de eventos. O listener de eventos então chama o serviço de e-mail externo.

Serviço de e-mail retorna um erro

O cenário descrito funciona muito bem se serviço de e-mail é operacional. No entanto, o que acontece se o serviço de e-mail Está retornando um erro? As solicitações do listener do evento ao serviço de e-mail falharão e, portanto, nenhuma mensagem de confirmação será enviada ao cliente. Como, no momento da falha, o evento de alteração de documento do Couchbase já foi processado, nenhum novo evento para o mesmo documento será iniciado a menos que haja outra alteração nele. Para garantir que a confirmação seja enviada, precisamos lidar com o erro e implementar um mecanismo de tentativa e erro. Ao fazer isso, podemos contornar qualquer problema temporário no serviço externo e, ao mesmo tempo, garantir que a confirmação seja enviada.

Existem diferentes maneiras de abordar isso. No meu exemplo abaixo, escolho criar uma nova coleção chamada ‘retry’, que armazenará referências aos documentos para os quais a execução do listener do evento falhou.

O listener de eventos receberá alterações na ordem dos documentos (passo #1) e, em seguida, chamará o serviço de e-mail (passo #2). Se a chamada ao serviço de e-mail for bem-sucedida, o listener de eventos atualizará o status da mensagem de confirmação no documento do pedido (passo #3). No entanto, em caso de falha, será criado um documento de tentativa novamente e colocado na coleção de tentativas (passo #3*).

Manter uma referência aos documentos permite-nos identificar todas as atualizações que falharam e permite-nos executá-las novamente posteriormente. Isso pode ser feito através de intervenção manual por parte de um operador ou através de tentativas automáticas utilizando temporizadores de eventos do Couchbase.

Iniciamos o processo de replicação adicionando um documento com o id do documento especificado à coleção de replicação. É criado um temporizador recorrente com um intervalo temporizador fornecido.
Na execução do temporizador, todos os documentos mais antigos que um pequeno período de tempo no sistema tentar novamente as coleções são atualizadas. Ao adicionar um atributo como fireRetry = true para os documentos de replicação, ativamos outro evento de atualização que é captado pelo listener do evento para executar o mecanismo de replicação. Isso nos proporciona uma mutação recursiva que ativa todos os documentos da coleção de replicação em paralelo. A função de replicação agora é executada usando todos os threads de trabalho disponíveis em paralelo.
Um evento de atualização de documento é acionado para cada documento retornado individualmente.
O documento de ordem correspondente é recuperado da coleção de entrada.
Agora o serviço de e-mail é chamado.
Se a chamada ao serviço de e-mail for bem-sucedida, o listener de eventos atualizará o status da mensagem de confirmação no documento do pedido e removerá o documento de tentativa e erro.
Em caso de falha, o documento de replicação é atualizado e inserido na coleção ‘retry’.

Revisão de código

Agora que temos o design conceitual estabelecido, vamos dar uma olhada na implementação de exemplo:

Pré-requisitos:

Couchbase 7 Enterprise Edition. Eu executarei o Couchbase como um cluster de um único nó no Docker na minha máquina local. (https://docs.couchbase.com/server/current/install/getting-started-docker.html)
Para fins de desenvolvimento, criamos um único cluster Couchbase, executando os seguintes serviços:
- Serviço de Indexação, Consulta, Eventos e Dados

Por favor, observe que a instalação de um único nó não é recomendada para uso em produção.

Preparação

Crie um bucket chamado pedidos
Crie duas coleções no pedidos bucket_default_scope:
- entrada (este arquivo conterá todas as ordens recebidas)
- tentar novamente (este arquivo conterá os documentos de replicação que referenciam as ordens que falharam)

Criar um bucket ‘meta-dados’. Usaremos o escopo _default e a coleção _default. O bucket de metadados é usado para os metadados do Eventing.

Crie um índice no tentar novamente Coleção. O listener de retorno de tentativas consulta todos os documentos contidos na coleção usando o N1QL, portanto é necessário ter um índice disponível para que a consulta seja executada.

CREATE PRIMARY INDEX idx_default_primary ON orders._default.retry USING GSI;

1	Criar PRIMEIRO ÍNDICE idx_default_primary LIGADO pedidos._default.tentar novamente USANDO GSI;

Documento de ordem do modelo de dados

Para este exemplo de aplicação, utilizamos um modelo de dados simples para o documento "Order", que contém apenas os campos relevantes. Muitos outros campos que você normalmente esperaria em um documento de pedido são omitidos.

{
  “email”: “customer_email”,
  “paymentStatus”: “initiated”,
  “confirmationEmailSent”: false,
  “items”: [
    {
      “name”: “Swedish Meatballs 500g”,
      “amount”: 2,
      “unitPrice”: 9.95
    }
  ]
}

{

“e-mail”: “customer_email”,

“paymentStatus”: “iniciada”,

“ConfirmationEmailSent”: falso,

“itens”: [

{

“nome”: “Bolas de carne suecas 500g”,

“quantidade”: 2,

“unitPrice”: 9.95

}

]

}

Documento de replicação do modelo de dados

O documento de replicação contém alguns atributos básicos, como o identificador do documento da ordem, um contador de tentativas e uma marca de tempo. tipo O atributo não é necessário no nosso aplicativo, mas pode ser útil para determinar o tipo de notificação por e-mail, caso o aplicativo seja expandido para enviar também informações sobre o envio e a entrega.

{
  “type”: “confirmation”,
  “docId”: “order_140”,
  “attempt”: 1,
  “ts”: 1632775908319
}

{

“tipo”: “confirmação”,

“docId”: “order_140”,

“tentativa”: 1,

“ts”: 1632775908319

}

Serviço de e-mail MOCK

Vamos burlar o serviço de e-mail usando um simples script Python que executa um servidor web local. O script responderá aleatoriamente com HTTP 200 OK ou HTTP 406 para indicar um erro.

Atualize o endereço IP para o endereço IP da sua máquina local na linha 31server = ThreadedHTTPServer((‘substitua com a sua IP’, 9080), Handler)
Inicie o script executando: python http.py

Os ouvintes de eventos

Agora, com todos os preparativos feitos, podemos continuar e adicionar os dois listeners de evento:

evt_send_confirmation_email – fornece a integração com o Serviço de E-mail
evt_send_confirmation_email_retry – contém a lógica de replicação de tentativas

Os ouvintes estão disponíveis aqui: https://github.com/puhhma/cb_eventing_retry_sample
Importe os ouvintes (json arquivo) para o serviço Couchbase Eventing.

Por favor, observe que para que os ouvintes consigam trabalhar, é necessário seguir as convenções de nomenclatura utilizadas neste artigo.

Avaliação evt_send_confirmation_email ouvidor

Configuração de listeners de eventos:

O listener de eventos está ouvindo o entrada coleta no local pedidos balde.
A meta-dados O bucket é usado para armazenar os metadados dos listeners
Os aliases de bucket bkt_order_inbound e bkt_order_retry refira-se à correspondente entrada e tentar novamente coleta no local pedido balde
A curlEmailServiceHost Especifica o alias URL do mock EmailService. Certifique-se de atualizar com o seu endereço IP.

// OnUpdate is invoked for all documents created/updated in the ‘inbound’ bucket
function OnUpdate(doc, meta) {
    // determine if document status is ‘paid’ & confirmation email was not previously sent
    if( doc.paymentStatus === “paid” && !doc.confirmationEmailSent ) {
        SendConfirmationMail(doc, meta.id);
    } else {
      if (debug_level > 1)
        log(“Nothing to do for: ” + meta.id);
    }
}

function SendConfirmationMail(doc, docId) {
  try {
      // build the request to the EmailService
      var request = {
          path: ‘sendConfirmation’,
          body: doc
      };
      //  perform the cURL request using the URL alias ‘curlEmailServiceHost’ from the settings
      var response = curl(‘POST’, curlEmailServiceHost, request);
      if (response.status != 200) {
          // this did not work as expected
          if (debug_level > 1) {
              log(“docId”, docId, “cURL POST failed response.status:”,response.status);
          }
          // create retry document referencing the documentId and store in ‘retry’ bucket
          bkt_order_retry[docId] = {
              “docId”: docId,
              “attempt”: 1,
              “ts”: Date.now()
          }
      } else {
          if (debug_level > 5) {
              log(“cURL POST success, sent”,docId,”response.body:”,response.body);
          }
          // update confirmationEmailSent status
          doc.confirmationEmailSent = true;
          bkt_order_inbound[docId] = doc;
      }
  } catch (e) {
      log(“ERROR cURL request had an exception:”,e)
  }
}

// O método OnUpdate é chamado para todos os documentos criados/atualizados na pasta ‘inbound’

função AoAtualizar(doc, meta) {

// determinar se o status do documento é ‘pago’ e se o e-mail de confirmação não foi enviado anteriormente

se( doc.statusDoPagamento === “pagado” && !doc.E-mail de confirmação enviado ) {

Enviar e-mail de confirmação(doc, meta.identidade);

} senão {

se (debug_level > 1)

log(“Não há nada a fazer: “ + meta.identidade);

}

função Enviar e-mail de confirmação(doc, idDoc) {

tentar {

// criar a solicitação para o EmailService

variável pedido = {

caminho: ‘sendConfirmation’,

corpo: doc

};

// Executar a solicitação cURL usando o alias URL ‘curlEmailServiceHost’ nas configurações

variável resposta = curl(‘POST’, curlEmailServiceHost, pedido);

se (resposta.status != 200) {

// isso não funcionou como esperado

se (debug_level > 1) {

log(“docId”, idDoc, “A resposta do comando cURL POST não foi recebida com sucesso. Status::”,resposta.status);

}

// criar um documento de replicação com referência ao ID do documento e armazená-lo no bucket ‘retry’

bkt_order_retry[idDoc] = {

“docId”: idDoc,

“tentativa”: 1,

“ts”: Data.Agora()

}

} senão {

se (debug_level > 5) {

log(“cURL POST sucesso, enviado”,idDoc,“response.body:”,resposta.corpo);

}

// status de confirmação de atualização do e-mail enviado

doc.E-mail de confirmação enviado = verdadeiro;

bkt_order_inbound[idDoc] = doc;

}

} capturar (e) {

log(“A solicitação do cURL apresentou uma exceção:”,e)

}

Consulte os comentários em linha para obter detalhes.
A AoAtualizar A função é ativada quando um documento de pedido é atualizado ou criado.
O pedido é construído e a solicitação HTTP POST é enviada para o EmailService usando o cURL.
O resultado é avaliado. Caso a resposta HTTP não seja bem-sucedida, um documento de replicação é construído e adicionado ao arquivo tentar novamente coleta.

O listener de eventos está ouvindo o tentar novamente coleta no local pedidos balde.
A meta-dados O bucket é usado para armazenar os metadados dos listeners
Os aliases de bucket bkt_order_inbound e bkt_order_retry refira-se à correspondente entrada e tentar novamente colheita na caixa de ordem
A curlEmailServiceHost Especifica o alias URL para o mock EmailService
A retryTimerIntervalo Especifica o intervalo do temporizador em segundos.

function OnUpdate(doc, meta) {
    if (meta.id === “allow_retrys”) {
        // the timer is initialized by creating document with id = ‘allow_retrys’
        CreateRetryTimer({“id”: meta.id, “mode”: “initial”});
    } else if (doc.fireRetry) {
      // process retry documents
      SendConfirmationMail(doc, meta.id);
    }
}

function CreateRetryTimer(context) {
    if (debug_level > 2) {
        log(‘From CreateRetryTimer: creating timer’, context.mode, context.id);
    }
    // Create a timestamp ‘retryTimerInterval’ seconds (from the settings) from now
    var timerStartTime = new Date();
    // Get current time & add ‘retryTimerInterval’ sec. to it.
    timerStartTime.setSeconds(timerStartTime.getSeconds() + retryTimerInterval);
    // Create a document to use as out for our context
    createTimer(RetryTimerCallback, timerStartTime, context.id, context);
}

function RetryTimerCallback(context) {
    if (debug_level > 2) {
        log(‘From RetryTimerCallback: timer fired’, context);
    }
    // rearm the timer ASAP, to ensure timer keeps running in the event
    // of later  errors or script timeouts in later “recurring work”.
    CreateRetryTimer({ “id”: context.id, “mode”: “via_callback” });
    // Update all retry documents in the ‘retry’ bucket. Exclude the ‘allow_retys’ document
    // and any documents that were created more than 15 seconds ago, in order to avoid retry ‘to early’.
    N1QL(“UPDATE orders._default.retry SET fireRetry = true WHERE meta().id != ‘allow_retrys’ AND ts < DATE_ADD_MILLIS(NOW_MILLIS(), -15, ‘second’)”);
}

function SendConfirmationMail(retryDoc, docId) {
  try {
      // resolve order document by id
      var doc = bkt_order_inbound[docId];
      // build the request
      var request = {
          path: ‘sendConfirmation’,
          body: doc
      };
      //  perform the cURL request using the URL alias from the settings
      var response = curl(‘POST’, curlEmailServiceHost, request);
      if (response.status != 200) {
          // this did not work as expected
          if (debug_level > 1) {
              log(“docId”, docId, “cURL POST failed response.status:”,response.status);
          }
        // increment attempt count in retry document
        retryDoc.attempt = ++retryDoc.attempt;
        // Set fireRetry = false, to avoid retry execution with this document change
        retryDoc.fireRetry = false;
        retryDoc.ts = Date.now();
        // update retry document
        bkt_order_retry[docId] = retryDoc;
      } else {
          if (debug_level > 5) {
              log(“cURL POST success, sent”,docId,”response.body:”,response.body);
          }
          doc.confirmationEmailSent = true;
          bkt_order_inbound[docId] = doc;
          // delete the retry document
          delete bkt_order_retry[docId];
      }  } catch (e) {
      log(“ERROR cURL request had an exception:”,e)
  }
}

função AoAtualizar(doc, meta) {

se (meta.identidade === “allow_retrys”) {

// o temporizador é inicializado criando um documento com o id = ‘allow_retrys’

CreateRetryTimer({“id”: meta.identidade, “mode”: “inicial”});

} senão se (doc.fireRetry) {

// documentos que retiram processos de tentativa e erro

Enviar e-mail de confirmação(doc, meta.identidade);

}

função CreateRetryTimer(contexto) {

se (debug_level > 2) {

log(‘De CreateRetryTimer: criando o timer’, contexto.modo, contexto.identidade);

}

// Crie um temporizador de repetição com o valor de segundos ‘retryTimerInterval’ (de acordo com as configurações atuais)

variável timerStartTime = novo Data();

// Obtenha a hora atual e adicione a variável ‘retryTimerInterval’ em segundos.

timerStartTime.setSeconds(timerStartTime.getSeconds() + retryTimerInterval);

// Crie um documento para usar como saída para nosso contexto

createTimer(RetryTimerCallback, timerStartTime, contexto.identidade, contexto);

}

função RetryTimerCallback(contexto) {

se (debug_level > 2) {

log(‘De RetryTimerCallback: o timer foi disparado’, contexto);

}

// Arme o temporizador o mais rápido possível, para garantir que ele continue funcionando no caso de ocorrer algum evento.

// de erros posteriores ou tempos de espera de script em trabalhos posteriores que se repetem.

CreateRetryTimer({ “id”: contexto.identidade, “mode”: “via_callback” });

// Atualizar todos os documentos de retries no bucket ‘retry’. Excluir o documento ‘allow_retys’

// e quaisquer documentos que tenham sido criados há mais de 15 segundos, a fim de evitar a repetição de processos que se iniciaram antes do tempo previsto.

N1QL(“UPDATE orders._default.retry SET fireRetry = true WHERE meta().id != ‘allow_retrys’ AND ts < DATE_ADD_MILLIS(NOW_MILLIS(), -15, ‘second’)”);

}

função Enviar e-mail de confirmação(retryDoc, idDoc) {

tentar {

// resolver a ordem de documentos por ID

variável doc = bkt_order_inbound[idDoc];

// criar o pedido

variável pedido = {

caminho: ‘sendConfirmation’,

corpo: doc

};

// executar a solicitação cURL usando o alias de URL dos ajustes

variável resposta = curl(‘POST’, curlEmailServiceHost, pedido);

se (resposta.status != 200) {

// isso não funcionou como esperado

se (debug_level > 1) {

log(“docId”, idDoc, “A resposta do comando cURL POST não foi recebida com sucesso. Status::”,resposta.status);

}

// incremento do número de tentativas no documento de replicação

retryDoc.tentativa = ++retryDoc.tentativa;

// Definir SetFireRetry = false, para evitar a execução de novas tentativas de execução com esta alteração no documento

retryDoc.fireRetry = falso;

retryDoc.ts = Data.Agora();

// atualização de tentativa de atualização do documento

bkt_order_retry[idDoc] = retryDoc;

} senão {

se (debug_level > 5) {

log(“cURL POST sucesso, enviado”,idDoc,“response.body:”,resposta.corpo);

}

doc.E-mail de confirmação enviado = verdadeiro;

bkt_order_inbound[idDoc] = doc;

// apagar o documento de tentativa de reinicialização

eliminar bkt_order_retry[idDoc];

} } capturar (e) {

log(“A solicitação do cURL apresentou uma exceção:”,e)

}

O temporizador é iniciado ao adicionar um documento com o id permitir retris para o tentar novamente coleção
O temporizador é então inicializado e a função RetryTimerCallback associada ao temporizador é chamada.
Assim que o temporizador for executado, a função RetryTimerCallback será chamada.
Antes de prosseguir com o mecanismo de replicação, é criado um novo temporizador como primeiro passo para garantir que ele continue funcionando em caso de erros posteriores.
Uma consulta N1QL é usada para atualizar todas as informações. tentar novamente documentos no tentar novamente Coleção adicionando um fireRetry Atribuir ao documento
Cada alteração de documento resulta em um evento de atualização de documento e o mecanismo de replicação é executado.
O documento de ordem é resolvido a partir do entrada A coleção e o serviço de e-mail são chamados via cURL
Em caso de falha, tentar novamente o documento é atualizado e tentativa O contador aumentou

Teste o aplicativo de demonstração

Agora é hora de testar finalmente o aplicativo de demonstração:

Certifique-se de que o mock EmailService está ativo e funcionando
Inicie o evt_send_confirmation_email listener de evento, mas mantenha o evt_send_confirmation_email_retry O ouvinte parou por enquanto.
Crie um documento de pedido de amostra (consulte o modelo de dados acima) no console da Couchbase
No caso de uma resposta positiva, E-mail de confirmação enviado O atributo é atualizado para true na pedido documento.
Em caso de falha, um documento de replicação é criado no tentar novamente coleta. Como o EmailService responderá aleatoriamente com um erro, por favor repita o passo #3 até que ocorra um erro.
Agora que conseguimos capturar um erro, vamos mesmo começar a ouvir o evento de tentativa novamente evt_send_confirmation_email_retry
Crie um documento com o id ‘allow_retrys’. Isso iniciará o mecanismo de retornos.
Após um curto período, o ouvinte se torna ativo e começa a processar os documentos apresentados. tentar novamente coleta.
Por favor, observe que o atributo ‘attempt’ aumenta a cada atualização falhada do Serviço de E-mail. No caso de um sucesso, o documento da ordem é atualizado e o correspondente tentar novamente o documento foi removido do tentar novamente coleta.

Como a resposta do Teste de Serviço de E-mail é aleatória, poderá ser necessário repetir os passos acima para observar o comportamento esperado.

Conclusão

Neste artigo, apresento um mecanismo de replicação para lidar com condições de erro ao integrar o Couchbase Eventing com um serviço REST externo. Estas ou soluções semelhantes podem ser utilizadas para garantir que as ações previstas sejam realizadas mesmo que o serviço externo esteja temporariamente com problemas.

Ao considerar um mecanismo de replicação, vários fatores precisam ser considerados, como o volume de replicações, o número de threads disponíveis para o Couchbase Eventing e as solicitações que o serviço externo pode lidar.

Você pode encontrar mais informações sobre o funcionamento interno do Couchbase Eventing aqui: https://docs.couchbase.com/server/current/eventing/eventing-overview.html

Muito obrigado a Jon Strabala (Gerente de Produto Principal, Couchbase) pelo insight técnico e pelo apoio neste artigo.

Por que Couchbase

Por setor

Por Caso de Uso

Centro de recursos

Construa com o Couchbase

Quem somos

Parceiros

Processamento de erros e tentativas de replicação no Couchbase Eventing

Migration Guide: How to Move from MongoDB® to Couchbase

Why Vector-Only RAG Fails on Analytical Questions, and a Reference Architecture That Doesn’t

Couchbase Named a Leader in the 2026 IDC MarketScape for Worldwide Operational Databases

Benchmarking Billion-Scale Vector Indexes

Por que os agentes de IA falham na produção: o problema da camada de dados

Deixe a proliferação de bancos de dados no passado

Por que Couchbase

Por setor

Por Caso de Uso

Centro de recursos

Construa com o Couchbase

Quem somos

Parceiros

Processamento de erros e tentativas de replicação no Couchbase Eventing

Receba as atualizações do blog do Couchbase na sua caixa de entrada

Autor

Publicado por Marian Puhl, Senior Solutions Engineer

Deixe um comentário Cancelar resposta

Deixe a proliferação de bancos de dados no passado