diff --git a/docs/content/docs/guides/clay.mdx b/docs/content/docs/guides/clay.mdx index a63da7b09..227d2f054 100644 --- a/docs/content/docs/guides/clay.mdx +++ b/docs/content/docs/guides/clay.mdx @@ -94,7 +94,7 @@ Clay can map the response into columns. A single-row request answers with that r `status` is `created`, `updated`, `skipped` (for example, already a contact with **Leave it alone** chosen, or suppressed) or `failed`. A skipped or failed row carries a `reason`. A request for many rows has the same counts and one entry per row in `results`. -A URL Warmbly does not know answers `404`, a disconnected connection `403`, and more than 600 requests a minute on one connection `429` with `Retry-After`. Clay retries those on its own. +A URL Warmbly does not know answers `404` and a disconnected connection `403`. Neither goes away on a retry: fix the URL in the column or reconnect Clay, then rerun the rows. More than 600 requests a minute on one connection answer `429` with `Retry-After`, and a Warmbly error answers `500`. Those are worth retrying, so turn on retries in the HTTP API column's settings and include `429` and `500` in the statuses it retries. ### What rows become @@ -134,7 +134,7 @@ Choose the events to send, save, and use **Send a test row** to check that Clay The events are: prospect replies, email sent, email opened, link clicked, email bounces, contact unsubscribes, meeting booked and meeting canceled. Each one adds a row with the event, the contact's email and name, the campaign, and the event's details as flat columns. -To send to more than one Clay table, use an [automation](/guides/automations/) with the **Send a row to Clay** action. It sends to the connection's table unless the step names another table webhook. +To send to more than one Clay table, use an [automation](/guides/automations/) with the **Send a row to Clay** action. It sends to the connection's table unless the step names another table webhook. The connection's auth token is only sent to its own table, so a table named on a step needs a webhook source without a token. **Stop sending** removes the webhook URL, the token and the chosen events. diff --git a/internal/api/handler/clay.go b/internal/api/handler/clay.go index 0abe94b76..cba8de206 100644 --- a/internal/api/handler/clay.go +++ b/internal/api/handler/clay.go @@ -113,7 +113,7 @@ func (h *Handler) InboundClay(c *gin.Context) { c.JSON(int(xe.Code), gin.H{"status": "failed", "error": xe.Message}) default: log.Error().Err(err).Msg("clay inbound failed") - c.JSON(http.StatusInternalServerError, gin.H{"status": "failed", "error": "Warmbly could not save these rows. Clay will retry."}) + c.JSON(http.StatusInternalServerError, gin.H{"status": "failed", "error": "Warmbly could not save these rows. Run them again."}) } return } diff --git a/internal/app/apollo/service.go b/internal/app/apollo/service.go index 5587bbb84..36ffdba3d 100644 --- a/internal/app/apollo/service.go +++ b/internal/app/apollo/service.go @@ -29,6 +29,10 @@ const ( RecurringEvery = time.Hour // enrichChunk is how many contacts an enrichment run loads and saves at once. enrichChunk = 100 + // runTimeout bounds one import run; staleAfter is when a run still marked + // running is treated as interrupted, and must stay above runTimeout. + runTimeout = 25 * time.Minute + staleAfter = 30 * time.Minute ) // Deps is everything the Apollo integration talks to. @@ -383,7 +387,7 @@ func (s *Service) StartRun(ctx context.Context, orgID, id uuid.UUID) (*models.Ap return nil, errx.NewWithIdentifier(errx.Conflict, "import_running", "This import is already running.") } go func(src models.ApolloImportSource) { - bg, cancel := context.WithTimeout(context.Background(), 25*time.Minute) + bg, cancel := context.WithTimeout(context.Background(), runTimeout) defer cancel() s.runSource(bg, src) }(*src) @@ -393,6 +397,9 @@ func (s *Service) StartRun(ctx context.Context, orgID, id uuid.UUID) (*models.Ap // RunRecurring runs every "keep in sync" source that is due. func (s *Service) RunRecurring(ctx context.Context) error { + if _, err := s.Repo.ReleaseStaleSources(ctx, staleAfter); err != nil { + log.Warn().Err(err).Msg("apollo: could not release interrupted imports") + } due, err := s.Repo.DueRecurringSources(ctx, RecurringEvery) if err != nil { return err @@ -405,7 +412,9 @@ func (s *Service) RunRecurring(ctx context.Context) error { if err != nil || !claimed { continue } - s.runSource(ctx, src) + run, cancel := context.WithTimeout(ctx, runTimeout) + s.runSource(run, src) + cancel() } return nil } @@ -422,7 +431,10 @@ func (s *Service) runSource(ctx context.Context, src models.ApolloImportSource) } log.Warn().Err(err).Str("source", src.ID.String()).Msg("apollo: import run failed") } - _ = s.Repo.FinishSource(ctx, src.ID, res, msg) + // The outcome is recorded even when the run's own context has ended. + if err := s.Repo.FinishSource(context.WithoutCancel(ctx), src.ID, res, msg); err != nil { + log.Warn().Err(err).Str("source", src.ID.String()).Msg("apollo: could not record import result") + } } // importOnce reads the list, imports people it has not brought in before and diff --git a/internal/app/integration/clay.go b/internal/app/integration/clay.go index 831060fba..815401785 100644 --- a/internal/app/integration/clay.go +++ b/internal/app/integration/clay.go @@ -101,9 +101,10 @@ var ErrConnectionNotFound = errors.New("connection not found") // clayDeliver posts one flat row to a Clay table's webhook source. func (s *service) clayDeliver(ctx context.Context, sub models.IntegrationEventSubscription, secretCfg map[string]any, data map[string]any, msg eventMessage) error { + saved := stringFromMap(secretCfg, "webhook_url") target := configString(sub.Config, "url") if target == "" { - target = stringFromMap(secretCfg, "webhook_url") + target = saved } if target == "" { return errors.New("add the Clay table's webhook URL on the Clay page first") @@ -112,8 +113,9 @@ func (s *service) clayDeliver(ctx context.Context, sub models.IntegrationEventSu if err != nil { return err } + // The connection's sealed token goes only to the connection's own table webhook. token := configString(sub.Config, "auth_token") - if token == "" { + if token == "" && target == saved { token = stringFromMap(secretCfg, "auth_token") } return clayPost(ctx, rendered, token, clayRow(sub.EventType, data, msg)) diff --git a/internal/repository/pg_apollo.go b/internal/repository/pg_apollo.go index 8894507c9..5b001c10b 100644 --- a/internal/repository/pg_apollo.go +++ b/internal/repository/pg_apollo.go @@ -40,6 +40,7 @@ type ApolloRepository interface { ClaimSource(ctx context.Context, id uuid.UUID) (bool, error) FinishSource(ctx context.Context, id uuid.UUID, result *models.ApolloRunResult, errMsg string) error DueRecurringSources(ctx context.Context, every time.Duration) ([]models.ApolloImportSource, error) + ReleaseStaleSources(ctx context.Context, after time.Duration) (int64, error) SourceMembers(ctx context.Context, sourceID uuid.UUID, apolloIDs []string) (map[string]bool, error) AddSourceMembers(ctx context.Context, sourceID uuid.UUID, members map[string]uuid.UUID) error @@ -180,6 +181,19 @@ func (r *apolloRepository) ClaimSource(ctx context.Context, id uuid.UUID) (bool, return tag.RowsAffected() > 0, nil } +// ReleaseStaleSources marks runs that outlived any live run (a restart mid-import) +// as interrupted, so they show an error, can be run again and sync on schedule. +func (r *apolloRepository) ReleaseStaleSources(ctx context.Context, after time.Duration) (int64, error) { + tag, err := r.db.Exec(ctx, ` + UPDATE apollo_import_sources + SET status = 'error', last_error = 'The last run was interrupted. It runs again on its next check, or choose Run now.', updated_at = NOW() + WHERE status = 'running' AND updated_at < NOW() - make_interval(secs => $1)`, after.Seconds()) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + func (r *apolloRepository) FinishSource(ctx context.Context, id uuid.UUID, result *models.ApolloRunResult, errMsg string) error { var raw []byte added := 0 @@ -206,7 +220,8 @@ func (r *apolloRepository) DueRecurringSources(ctx context.Context, every time.D s.last_result, s.last_error, s.total_imported, s.created_at, s.updated_at FROM apollo_import_sources s JOIN integration_connections c ON c.id = s.connection_id - WHERE s.recurring AND s.enabled AND s.status <> 'running' + WHERE s.recurring AND s.enabled + AND (s.status <> 'running' OR s.updated_at < NOW() - INTERVAL '30 minutes') AND c.status IN ('connected', 'degraded') AND (s.last_run_at IS NULL OR s.last_run_at < NOW() - make_interval(secs => $1)) ORDER BY s.last_run_at NULLS FIRST diff --git a/web/src/app/app/integrations/_components/ConnectionDetail.tsx b/web/src/app/app/integrations/_components/ConnectionDetail.tsx index 9d520d370..cd877e836 100644 --- a/web/src/app/app/integrations/_components/ConnectionDetail.tsx +++ b/web/src/app/app/integrations/_components/ConnectionDetail.tsx @@ -227,36 +227,36 @@ export default function ConnectionDetail({ {/* Salesforce has its own page: sync rules, field map, imports, activity log. */} {conn.provider === "salesforce" && (
Set this to send to a different Clay table than the connection default.
+Sends to a different Clay table than the connection default. The connection's auth token is not sent to it.