ci/python-sdk-e2e-sqlite
38 Commits
| Author | SHA1 | Message | Date | |
|---|---|---|---|---|
 Manan Bhatt
|
69c16eae27 |
fix(persistence): index end_time so End Time can be sorted and filtered
The End Time column on the executions and task search pages is marked sortable, and the date-range picker emits `endTime>`/`endTime<` clauses. `end_time` was never a column on workflow_index or task_index, so the query builders snake_cased the field, failed to match it against VALID_FIELDS, and dropped it -- from the ORDER BY via getSort() and from the WHERE via Condition.isValid(). Sorting by End Time returned an unordered page and lost the default `startTime:DESC`; filtering by an End Time range silently returned unfiltered results. Add end_time to both index tables on sqlite and postgres, index it, populate it from the summary on write, and add it to the allow-list. The millis-to-UTC conversion applied to filter values keys off the `_time` suffix, so the date range needs no further change. The column is NOT NULL defaulting to the epoch rather than nullable. End time is absent until an execution is terminal, and SQLite sorts NULL lowest while Postgres sorts it highest, so a nullable column would put still-running executions at opposite ends of an endTime:DESC page depending on the backend. getSort() emits a bare `attribute + " " + order` with no room for a NULLS clause, so the sentinel is what keeps the two engines agreeing: unfinished rows carry the epoch and sort last on DESC, first on ASC, on both. This follows the update_time sentinel already in V1 and V13.1. end_time is added to the ON CONFLICT update clause as well as the insert, since the row already exists by the time a workflow finishes. Existing rows are back-filled from json_data, which already holds the instant as an ISO-8601 string. Terminal executions are never re-indexed, so without a back-fill every historical execution would sort as unfinished forever. Both back-fills are written to survive a malformed value rather than abort. Sqlite's strftime() returns NULL on one and COALESCE keeps the epoch. Postgres needs more care than V13.2 took: to_timestamp with a fixed 'YYYY-MM-DD"T"HH24:MI:SS.MS' raises `invalid value "Z" for "MS"` on an endTime with no milliseconds, and since applyDataMigrations defaults to true that would abort the migration and stop the server booting. V18.1 uses a ::timestamptz cast, which accepts any ISO-8601 form and reads the trailing Z as UTC, behind a regex guard so a non-timestamp value skips the row instead of failing the statement. Verified on a 3913-workflow / 45531-task sqlite database: all 3822 finished workflows match json_data exactly with 0 mismatches, the 91 unfinished rows keep the epoch, and `ORDER BY end_time DESC` plans as `SCAN workflow_index USING INDEX workflow_index_end_time_idx`. On postgres, V18 then V18.1 against seeded data converts values with and without milliseconds and with offsets, and leaves empty, malformed and absent endTimes at the epoch. No UI change: only `workflowType` was ever remapped on the way out, so `endTime` already reached the server unchanged. |
||
|
Miguel Prieto
|
dc358c4a1d | fix: write and compare sqlite index timestamps in UTC (#1497) (#1498) | ||
|
Kowser
|
4d941c7982 |
fix(persistence): baseline-0 on main flyway so boot survives scheduler-first migration
- mirrors postgres scheduler config (@Qualifier postgresRetryTemplate) - smoke-test fixture gets a mysqlRetryTemplate bean alongside postgresRetryTemplate - scheduler flyway shares the schema (own history table, baselineVersion 0); main flyway must tolerate a non-empty schema the same way or boot depends on bean-init order - postgres: baselineVersion 0 (default 1 would silently skip V1__initial_schema if second) - sqlite: baselineOnMigrate + baselineVersion 0 (had neither) - trade-off: legacy DBs w/ tables but no history now fail on V1 "already exists" instead of the baseline hint; same operator remedy Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> |
||
|
Viren Baraiya
|
a49990eb85 | more fixes | ||
|
Viren Baraiya
|
c85fa0fb4f | support for agent task and other UI fixes | ||
|
Viren Baraiya
|
978c47a9f0 |
Move SecretsDAO to org.conductoross package and simplify agentspan integration
Relocates SecretsDAO out of the legacy com.netflix.conductor.dao package, adapts SkillMetadataDAO implementations accordingly, and removes the AgentSpan embedded environment post-processor, principal filter, and env-backed credential store in favor of a simpler configuration wired directly through application.properties (agentspan.embedded). |
||
|
bradyyie
|
0a5abcacb5 |
Add workflow classifier support for definitions and execution search
Backports the Orkes workflow classifier concept to OSS. The classifier is a tag derived from WorkflowDef.metadata (explicit metadata.classifier wins, else 'agent' when the AgentSpan stamp agent_sdk/agentDef is present, else 'workflow' for plain untagged defs). - common: new WorkflowClassifier utility; WorkflowSummary gains a classifier field (proto field 23) populated at index time - rest: GET /metadata/workflow supports ?classifier= filtering; workflow /search and /search-v2 support &classifier=, folded into the structured query as 'classifier IN (...)' - postgres/sqlite: classifier column on workflow_index (Flyway V17 / V5), written on index and filterable via the query builders; the 'workflow' token also matches pre-migration NULL rows - es7/es8/os2/os3: classifier keyword mapping for workflow documents |
||
|
Dale Brady
|
bdf268eeec | feat: Embed AgentSpan agents into OSS Conductor (runtime + UI), gated by conductor.integrations.ai.enabled (#1213) | ||
|
Naomi Most
|
797126ee0f |
Revert "feat(webhooks-oss): WAIT_FOR_WEBHOOK end-to-end — port + persistence …" (#1202)
This reverts commit
|
||
|
Naomi Most
|
227f0e24c5 |
feat(webhooks-oss): WAIT_FOR_WEBHOOK end-to-end — port + persistence across 5 backings (#1106)
* feat: port webhooks-oss module from orkes-conductor
Lifts the new webhooks-oss/ module created by orkes-io/orkes-conductor#3612
(scheduler-style OSS/enterprise split) into conductor-oss. All io.orkes.conductor.*
packages translated to org.conductoross.conductor.* per repo convention.
Replaces the prior webhook-task/ module (deleted on main, only stale build/
remnants existed) with a structurally faithful copy of webhooks-oss from
orkes-conductor.
Ported
- 17 main + 7 test files comprising the webhooks-oss module
(verifiers, hashing, IncomingWebhookService, WebhookTaskMapper,
WebhookWorkerProperties, etc.)
- 10 supporting classes (EventMessage, ErrorList, Tag, WebhookConfig,
IncomingWebhookEvent, WebhookExecutionHistory, WebhookDAO,
WebhookTaskService, TargetWorkflowCollector, TimeBasedUUIDGenerator)
into common/ and core/
Code rewrites beyond mechanical namespace translation
1. ApplicationException -> NonTransientException (2 sites in
IncomingWebhookService, 3 sites in TimeBasedUUIDGenerator).
OSS retired ApplicationException.
2. TimeBasedUUIDGenerator stripped of multi-tenant OrkesRequestContext
lookups. generate() still produces real time-based UUIDs via log4j
UuidUtil. generate(long) (test-only, unused in OSS) falls back to
UUID.randomUUID() with TODO. getOrgId() returns "_".
3. EventMessage: removed orgId field and List<ExtendedEventExecution>
eventExecutions field (transitive class not ported; orgId is tenant
context).
4. IncomingWebhookService: replaced two executionDAOFacade.addEventMessage()
calls with log.warn(). OSS's ExecutionDAOFacade has no addEventMessage
(Orkes-only audit feature). Rejected events still observable via logs;
audit-table persistence is a future enhancement.
5. Deleted 3 postgres DAO tests (AbstractTestDAO, PostgresDAOTestUtil,
PostgresWebhookTaskServiceTest). They exercise a PostgresWebhookTaskService
impl that has not been ported yet.
6. Inlined FeatureFlags.SECURITY as literal "conductor.security.enabled"
in 5 verifier tests. Avoided porting a 20-line constants class for one
string reference.
7. TargetWorkflowCollectorTest: org.apache.groovy.util.Maps.of(...) ->
java.util.Map.of(...). Same semantics.
8. WebhookTaskMapper relocated from com.netflix.conductor.core.execution.mapper
to org.conductoross.conductor.tasks.webhook (per repo namespace
convention). Added explicit imports for TaskMapper and TaskMapperContext.
9. webhooks-oss/build.gradle adds commons-lang3, spring-web (test), and
spring-boot-starter-web (compileOnly) for the at-compile-time imports.
Wiring
- settings.gradle: include 'webhooks-oss'
- server/build.gradle: implementation project(':conductor-webhooks-oss')
Verified
- ./gradlew :conductor-webhooks-oss:compileJava + compileTestJava: clean
- ./gradlew :conductor-webhooks-oss:test: all green
- ./gradlew :conductor-server:compileJava: clean (full server tree)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
* feat(webhooks-oss): in-memory DAO + REST endpoint for first end-to-end webhook
Adds the runtime pieces needed to receive a webhook over HTTP and store it
in the queue for later dispatch. The dispatch worker and config-CRUD API
land in the next iteration.
Added
- InMemoryWebhookDAO: lifted from the prior feat/webhook-foundation
attempt, extended to satisfy the richer WebhookDAO interface ported
from orkes. Annotated @Component @ConditionalOnMissingBean(name=
"webhookDAO") so a persistent impl can override.
- InMemoryWebhookTaskService: lifted similarly, with @Component
@ConditionalOnMissingBean(name="webhookTaskService").
- IncomingWebhookResource: tenant-free OSS port of the
webhooks-enterprise REST controller. POST /webhook/{id} and GET
/webhook/{id}. ~60 lines.
Code rewrite
- IncomingWebhookService: swapped constructor-injected TimeBasedUUIDGenerator
for IDGenerator (already a @Bean in ConductorCoreConfiguration). Removes
the @ConditionalOnProperty wiring trap and avoids needing
conductor.id.generator=time_based. Webhook event IDs are now random UUIDs
rather than time-based UUIDs, which is fine for the queue path that uses
them.
Matchers note
InMemoryWebhookDAO.createMatchers/getMatchers/removeMatchers are no-ops.
Orkes' impl computes matchers by reading WorkflowDefs and extracting
inputParameters.matches from WAIT_FOR_WEBHOOK tasks, which requires
MetadataDAO injection. Inert until a worker consumes them, so deferred
with an inline comment.
Verified
- :conductor-webhooks-oss:compileJava + compileTestJava: clean
- :conductor-webhooks-oss:test: all green
- :conductor-server:compileJava: clean
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
* feat(webhooks-oss): WebhookConfigService + WebhookConfigResource for CRUD
Adds the management surface so webhook configs can be registered before
they receive events. Combined with the previous commit's REST receive
endpoint, this is the full register-then-receive arc.
Added
- WebhookConfigService: tenant-free OSS port of orkes' enterprise
service. CRUD over WebhookDAO, calls TargetWorkflowCollector to
populate matchers, sanitizes secrets on read. No audit logging
(AuditUtils is enterprise-only); no FeatureFlags conditional
(always enabled in OSS).
- WebhookConfigResource: REST controller mounted at
/api/metadata/webhook (mirrors orkes path). POST/PUT/GET/DELETE +
GET-all. No security annotations (@PreAuthorize / @PostFilter are
enterprise-only); no tag endpoints (TagsService is enterprise).
Validation rules preserved.
Code rewrites vs orkes
- ApplicationException(CONFLICT, ...) -> ConflictException
- ApplicationException(NOT_FOUND, ...) -> NotFoundException
- ApplicationException(INVALID_INPUT, ...) -> NonTransientException
- TimeBasedUUIDGenerator -> IDGenerator (consistent with prior commit)
- Dropped AuditUtils, FeatureFlags.WEBHOOKS conditional, all security
annotations, all tag endpoints
- Dropped createdBy = OrkesAuthentication.getAuthenticatedUser().getId()
since OSS has no authenticated-user concept here
Verified
- :conductor-webhooks-oss:compileJava + compileTestJava: clean
- :conductor-webhooks-oss:test: all green
- :conductor-server:compileJava: clean
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
* feat(webhooks-oss): WebhookWorker + real matchers for end-to-end dispatch
Completes the end-to-end webhook arc. An OSS server can now:
1. Register a webhook config (POST /api/metadata/webhook)
2. Receive an incoming webhook (POST /webhook/{id})
3. Verify the signature
4. Store the event + enqueue
5. The worker polls the queue, matches against waiting WAIT_FOR_WEBHOOK
tasks via the hash service, completes those tasks, and starts any
workflows declared in workflowsToStart.
Added
- WebhookWorker: OSS port of the Orkes Enterprise WebhookWorker. Plain
@Component with @PostConstruct-started ScheduledExecutorService and
@PreDestroy shutdown (no LifecycleAwareComponent base, no
MetricsCollector / Monitors, no OrkesRequestContext orgId switching,
no ExtendedEventExecution audit bookkeeping). Failures log instead
of persist. ~250 lines.
Changes
- InMemoryWebhookDAO.createMatchers now actually computes matchers:
injects MetadataDAO, reads WorkflowDef by name+version, finds
WAIT_FOR_WEBHOOK / WAIT tasks, extracts inputParameters.matches and
keys them by workflowName;version;taskRef. Mirrors orkes'
PostgresWebhookDAO.createMatchers minus the orgId prefix and
workflowDefToUpdateMap cache.
Code rewrites vs orkes WebhookWorker
- LifecycleAwareComponent -> plain @Component with @PostConstruct /
@PreDestroy
- MetricsCollector, Monitors.error -> log statements
- OrkesRequestContext.setOrgId per message -> dropped
- recordEventExecution(...) (writes ExtendedEventExecution audit
records via ExecutionDAOFacade.updateEventExecution) -> log
statements. ExtendedEventExecution / ExtendedEventHandler are not
ported to OSS.
- executionDAOFacade.addEventMessage(...) -> dropped (method doesn't
exist in OSS; same call previously stubbed in IncomingWebhookService)
- orkesRedisExecutionDAO.getTask(taskId) -> executionDAOFacade.getTaskModel(taskId)
- WorkflowConsistency.DURABLE -> dropped (enum doesn't exist in OSS)
- ApplicationException -> dropped (we already log + throw via log line)
Verified
- :conductor-webhooks-oss:compileJava + compileTestJava: clean
- :conductor-webhooks-oss:test: green
- :conductor-server:compileJava: clean
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
* test(webhooks-oss): cover OSS-added pieces (62 tests, 0 failures)
Adds unit-test coverage for the four OSS-original classes added in
prior commits on this branch. None of these had tests previously.
Added
- WebhookConfigServiceTest (8 tests): id generation, conflict on
existing id, fresh-id passthrough, secret sanitization on read,
redacted-secret preservation on update, real-secret overwrite,
TargetWorkflowCollector delegation, removeWebhook ordering.
- InMemoryWebhookDAOTest (8 tests): CRUD for configs and events,
plus the matchers computation logic added in the previous commit:
null override stores empty, missing WorkflowDef skipped, real
WAIT_FOR_WEBHOOK task with matches stored under
workflowName;version;taskRef, task without matches skipped,
non-webhook task skipped, removeMatchers drops.
- WebhookConfigResourceTest (8 tests): validation (missing all three
targets, HEADER_BASED without headers), valid delegation, path-id
override on update, 404 on missing get/delete, secret sanitization
on get.
- WebhookWorkerTest (6 tests): event-not-found and config-not-found
early returns, workflowsToStart invokes WorkflowExecutor with the
right StartWorkflowInput (name, version, input merge, event name,
createdBy), non-integer version skipped, matcher hit completes the
waiting WAIT_FOR_WEBHOOK task and removes it from the task service,
terminal task is not re-completed.
Change
- WebhookWorker.handleMessage made package-private so the test can
invoke it directly without driving the ScheduledExecutorService.
Verified
- :conductor-webhooks-oss:test passes 62 tests, 0 failures
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
* test(webhooks-oss): WebhookTest for the put() delegation
Closes the last untested link in the integration chain. When a workflow
reaches a WAIT_FOR_WEBHOOK task, the Webhook WorkflowSystemTask's
start() must register the task with WebhookTaskService so the worker
can find it via hash later.
- WebhookTest (1 test): verifies start() calls
webhookTaskService.put(taskModel, workflow.getWorkflowVersion()) and
sets status to IN_PROGRESS.
Verified: :conductor-webhooks-oss:test 63/63 pass.
* test(webhooks-oss): end-to-end integration test wiring real beans
Exercises the full webhooks-oss bean graph end-to-end with only the
WorkflowExecutor, QueueDAO, ExecutionDAOFacade, MetadataDAO, and
ParametersUtils mocked. Validates the integration story without
needing a full @SpringBootTest (which would require a circular dep
on test-harness).
Added
- WebhooksOssEndToEndTest (4 tests):
- register_then_receive_storesEventAndPushesQueue: full POST path
from REST controller through config DAO → IncomingWebhookService
→ verification → event store → queue push.
- receive_unknownWebhookId_returnsNullAndNoEnqueue: idempotent-safe
drop when no config is registered.
- register_workflow_completion_path_end_to_end: the load-bearing
one. Wires WorkflowDef → matchers computation → system-task
put() → receive → worker.handleMessage → workflowExecutor.updateTask
with status=COMPLETED. Validates the full task-completion chain.
- receive_failedVerification_throwsAndNoEnqueue: verifier rejects,
NonTransientException thrown, no queue push.
Wires real instances of InMemoryWebhookDAO, InMemoryWebhookTaskService,
TargetWorkflowCollector, WebhookConfigService, IncomingWebhookService,
IncomingWebhookResource, WebhookHashingService, WebhookWorker, and
Webhook. Uses a stub verifier that rejects bodies starting with
"REJECT" to drive the failure path deterministically.
Total tests on this branch now 67 across 9 test classes, 0 failures.
* fix(webhooks-oss): four real bugs found by smoke test against running server
Started server-lite, registered a webhook config, delivered an HMAC-signed
event, watched it dispatch into a workflow that completed. Found and fixed
four issues along the way.
Bug 1: REST routes missing /api prefix
IncomingWebhookResource mounted at /webhook; OSS convention is /api/...
(see RequestMappingConstants.API_PREFIX). Without the prefix, requests
fell through to the static resource handler and returned 500. Fixed:
/webhook -> /api/webhook.
Bug 2: @PathVariable name inference
Bare @PathVariable String id failed at runtime with "parameter name
information not available via reflection. Ensure that the compiler
uses the '-parameters' flag." OSS controllers explicitly name each
binding (see e.g. MetadataResource). Fixed: all @PathVariable in
IncomingWebhookResource and WebhookConfigResource now use explicit
("id") names.
Bug 3: Mutation of stored WebhookConfig on read
WebhookConfigService.getWebhooks() and WebhookConfigResource.getWebhook()
mutated the returned config's secretValue in place to redact it. With
the in-memory DAO that returns shared references, this corrupted the
persisted secret. Next webhook delivery read secretValue=*** and HMAC
verification failed with "Illegal base64 character 2a" trying to decode
the *** placeholder. Fix: WebhookConfig gained toBuilder=true; both
sites now clone-then-redact instead of mutating. Added regression
assertions in WebhookConfigServiceTest and WebhookConfigResourceTest.
Bug 4: server-lite did not depend on webhooks-oss
Only server/build.gradle pulled in :conductor-webhooks-oss. server-lite
is the lighter local-dev target; users running it via :conductor-server-lite:bootRun
would not get the webhooks module loaded at all. Fixed: added the dep
alongside the other system tasks (http-task, json-jq-task, kafka).
Verified end-to-end against running server
1. POST /api/metadata/webhook -> 200, returns config with generated id
2. GET /api/metadata/webhook/{id} -> 200, secret redacted to ***
3. POST /api/metadata/workflow -> 200, register wf-smoke v1
4. POST /api/webhook/{id} with HMAC-SHA256 signed body -> 200
5. WebhookWorker dequeued the event, invoked WorkflowExecutor.startWorkflow,
wf-smoke v1 ran to COMPLETED in 655ms
Tests: 68 across 9 classes, 0 failures.
* fix(webhooks-oss): round 1 ruthless cleanup — license headers, contract bugs, dead code
Adversarial review surfaced three blockers before this PR can flip from
draft to reviewable. All addressed in this commit.
License headers — would have failed spotlessJavaCheck on first CI build
42 files reformatted by spotlessApply. Every "Copyright 2022 Orkes, Inc."
+ "Orkes Enterprise License" header now Apache 2.0 / Conductor Authors.
TimeBasedUUIDGenerator had no header at all — fixed. Import ordering
normalized across the affected modules per repo Spotless config.
WebhookConfigService.updateWebhook NPE on missing id
PUT /api/metadata/webhook/{id} with an id that doesn't exist would
dereference null and 500 instead of 404. Added an explicit
NotFoundException at the top of the method, matching the pattern in
WebhookConfigResource.getWebhook + deleteWebhook. Regression test added.
IncomingWebhookService.handleWebhook 200/null masquerade for unknown id
POST /api/webhook/{id} to an unregistered webhook id returned HTTP 200
with empty body — looked successful to the caller. Now throws
NotFoundException so the global exception mapper returns 404. Updated
WebhooksOssEndToEndTest accordingly.
Dead code
- IncomingWebhookService no longer builds EventMessage objects that
were discarded immediately (left over from the addEventMessage ->
log.warn rewrite). 28 lines removed.
- Dropped unused ExecutionDAOFacade dependency — constructor down from
5 to 4 args.
- Dropped unused EventMessage / DEAD_LETTER_QUEUE imports.
Verified
- :conductor-webhooks-oss:compileJava + compileTestJava: clean
- :conductor-webhooks-oss:spotlessCheck: clean
- :conductor-webhooks-oss:test: 69 tests / 0 failures
* fix(webhooks-oss): round 2 ruthless cleanup — should-fix items from adversarial review
Addresses the should-fix items surfaced by the pre-review adversarial pass.
WebhookTaskMapper relocated for OSS convention
org.conductoross.conductor.tasks.webhook.WebhookTaskMapper ->
org.conductoross.conductor.webhook.tasks.mapper.WebhookTaskMapper
Matches the AI module's precedent (org.conductoross.conductor.ai.tasks.mapper.*)
and collocates the mapper with the rest of the webhook code instead of
splitting it across two top-level packages.
Dead code in TimeBasedUUIDGenerator
Dropped generate(long) and getOrgId(String) — both had TODO stubs with
no OSS callers (verified by grep). Net: 16 lines removed, generate()
gets an @Override annotation, three "OSS-OrkesRequestContext-removed"
TODOs gone. The remaining methods (generate() and getDate(String)) are
the only ones reachable in OSS.
Build.gradle hygiene
- compileOnly 'spring-boot-starter-web' -> implementation: the module's
REST controllers use @RestController/@RequestMapping etc. at runtime,
not just compile time. compileOnly only worked because :conductor-server
happens to pull starter-web transitively; module would be non-functional
if consumed standalone.
- Same upgrade for spring-boot-autoconfigure (used by
@ConditionalOnMissingBean).
- stripe-java 29.4.0 and sendgrid-java 4.10.2 hardcoded -> bound to new
revStripe and revSendgrid in dependencies.gradle. Added a PINNED
comment explaining they're webhook-verifier-specific clients with no
shared rev.
- Dropped now-orphan test deps: spring-web (no longer needed after
IncomingWebhookService dropped HttpHeaders test usage indirectly),
postgres-persistence, testcontainers (postgresql + base), HikariCP,
duplicate spring-retry. The DAO tests these supported moved to
postgres-persistence in the parent split PR.
WebhookWorker.recordHistory off-by-one
if (hist.size() > lastRunWorkflowIdSize) -> >=. Previous form let the
list grow to N+1 before trimming, contradicting the property name.
Missing test coverage
- IncomingWebhookResourceTest (2 tests): delegation + return value
passthrough for both handleWebhook and handlePing endpoints.
- InMemoryWebhookTaskServiceTest (6 tests): put/get hash semantics,
missing matches throws, bucketing of multiple tasks at the same hash,
remove preserves vs drops bucket, unknown hash returns empty.
WebhookConfigServiceTest extended with the NotFoundException-on-missing-id
regression (mirrors the contract fix in round 1).
Verified
- :conductor-webhooks-oss:compileJava + compileTestJava clean
- :conductor-webhooks-oss:spotlessCheck clean
- :conductor-webhooks-oss:test 76 tests / 0 failures
- :conductor-server:compileJava clean
* chore: revert accidentally-committed unrelated files from prior session cruft
* fix(webhooks-oss): correctness — DLQ-friendly worker + matcher recomputation
Two production-impacting bugs from the adversarial review.
WebhookWorker.pollAndExecute no longer acks on failure
Previously a try/finally acked unconditionally, so any exception in
handleMessage (DB blip, OOM, NPE) silently dropped the webhook event.
The TODO comment "retries are not yet modelled" was not a fix.
Now: ack only after handleMessage returns normally. On failure, log
and let the queue's unack timeout redeliver. Poison messages still
get retried, but they hit the underlying QueueDAO impl's retry policy
(e.g. postgres queue's max-retry → dead-letter) rather than vanishing.
InMemoryWebhookDAO.createMatchers no longer caches stale criteria
Previously createMatchers walked the WorkflowDefs at registration
time and stored the extracted `matches` criteria. If a workflow def
was later updated to change its WAIT_FOR_WEBHOOK task's matches
inputParameter, getMatchers kept returning the old criteria until
someone re-PUT the webhook config.
Now: createMatchers stores only the receiverWorkflowNamesToVersions
*targets* (which DO need to be captured at registration time so the
expression-based override is preserved). getMatchers looks up the
WorkflowDefs from MetadataDAO and extracts matches on every call.
Slightly more expensive (one MetadataDAO read per workflow per webhook
event), but always correct.
Tests
- InMemoryWebhookDAOTest gains getMatchers_reflectsWorkflowDefUpdates_noStaleCache:
register matchers, assert criteria, swap the mock to return an updated
WorkflowDef, assert getMatchers returns the new criteria without
re-running createMatchers. Pins the regression.
- 77 tests / 0 failures.
* test(webhooks-oss): rewrite WebhookWorkerTest with real beans + cover pollAndExecute
Per AGENTS.md preference: use real implementations over mocks. The prior
WebhookWorkerTest was 100% mocks. Rewrote to wire real instances of
InMemoryWebhookDAO, InMemoryWebhookTaskService, WebhookHashingService, and
TargetWorkflowCollector. Only the deep infra is still mocked: QueueDAO,
WorkflowExecutor, ExecutionDAOFacade, MetadataDAO, ParametersUtils.
The full register→receive→dispatch flow remains covered by
WebhooksOssEndToEndTest. This class now focuses on worker-internal
semantics that don't surface end-to-end.
Added pollAndExecute coverage
Made pollAndExecute() package-private (was private — same change pattern
already applied to handleMessage for the same reason). Four new tests:
- pollAndExecute_emptyBatch_noop: empty pop, no ack called.
- pollAndExecute_success_acks: happy path acks the message after
handleMessage returns cleanly.
- pollAndExecute_handleMessageThrows_doesNotAck: REGRESSION TEST. Prior
impl had a try/finally that acked even on failure, silently dropping
webhook events on any exception. Test uses an anonymous InMemoryWebhookDAO
subclass that throws from getWebhookEvent to simulate a DB blip.
- pollAndExecute_mixedBatch_acksOnlySuccesses: batch with one good +
one poison message. Only the good one gets acked; the poison one is
left for the queue's unack timeout to redeliver.
Existing handleMessage tests rewired with real beans
- handleMessage_eventNotFound_returnsEarly: stores nothing, calls
handleMessage, expects no executor invocation.
- handleMessage_configNotFound_returnsEarly: stores event but not config;
asserts the event is NOT removed (so retry can happen).
- handleMessage_workflowsToStart_invokesExecutor: registers config with
workflowsToStart, asserts StartWorkflowInput contents (name, version,
input merge, event name) and that the event is removed after success.
- handleMessage_workflowsToStart_nonIntegerVersion_skipped: version is a
string, no executor call.
- handleMessage_matcherHit_completesWaitingTask: registers config + uses
real WebhookTaskService.put to register a waiting task; asserts that
webhookTaskService.get returns empty after handleMessage completes.
- handleMessage_matcherHit_terminalTask_skipped: pre-COMPLETED task is
not re-completed.
Total: 81 tests across 11 classes, 0 failures.
* docs(webhooks-oss): WAIT_FOR_WEBHOOK task + REST surface
Per project CLAUDE.md, new user-facing surfaces need source-derived docs.
The webhooks-oss module added /api/metadata/webhook (config CRUD) and
/api/webhook/{id} (event receive) without any documentation; users had no
on-ramp.
Adds docs/.../systemtasks/wait-for-webhook-task.md following the existing
system task doc pattern (wait-task.md template). Contents:
- Task type + how the dispatch chain works (mapper, hash, worker)
- inputParameters table (matches)
- WebhookConfig registration: endpoint, body, verifier types table
(HMAC_BASED, SIGNATURE_BASED, HEADER_BASED, SLACK_BASED, STRIPE,
TWITTER, SENDGRID — pulled from the actual Verifier enum)
- Delivery endpoint: status codes, response semantics
- Verified end-to-end curl example: same flow exercised during the
smoke test (register config, register workflow def, deliver HMAC-signed
event)
- conductor.webhook.worker.* properties table (pulled from
WebhookWorkerProperties.java)
- Failure semantics: verification failure / worker dispatch failure /
rejected event logging — matches the round-3 ack-only-on-success
worker change.
Registered in mkdocs.yml nav alongside wait-task.md.
* docs(webhooks-oss): restore Javadoc on WEBHOOK_QUEUE constant for Orkes parity
* fix(webhooks-oss): log verification failures at ERROR to match Orkes
* test(webhooks-oss): property test for registration/inbound hash agreement
Random (workflow, version, taskRef, matches, body) tuples assert the two
hash sites in webhooks-oss canonicalize identically: InMemoryWebhookTaskService
(registration) and WebhookHashingService.computeJsonHash (inbound). Catches
silent divergence that would cause inbound events to miss registered tasks.
100 reps per direction (match / no-match). Seed defaults to System.nanoTime();
on failure the TestWatcher prints -Dwebhook.hash.seed=<n> for reproducibility.
* fix(webhooks-oss): guard StripeVerifier against null api_version
Stripe events whose body omits api_version (older API payloads, synthetic
test payloads, and apparently some real Stripe deliveries) NPE on
rawApiVersion.split(...). The signature has already been verified at this
point, and both branches of the apiVersion check ultimately return
ErrorList.empty(), so a missing api_version is treated as "old api / no
deserializer support" — same as the dated pre-deserializer path.
Surfaced by the smoke harness firing synthetic Stripe-shaped bodies; saw
500 NPE in the verifier path. Adds StripeVerifierTest (3 cases): null
api_version accepted, bad signature rejected, missing header rejected.
Orkes has identical code in io.orkes.conductor.webhook.verifier.StripeVerifier
— mirror fix needed upstream to keep parity.
* feat(postgres-persistence): PostgresWebhookDAO + PostgresWebhookTaskService
Production multi-node backing for the WebhookDAO + WebhookTaskService
interfaces. Designed clean for OSS (no orgId anywhere in the schema or
queries) and applies the matcher-cache fix from the parent PR's Round 3:
target workflows are persisted, match criteria are recomputed from
MetadataDAO on every getMatchers() call so WorkflowDef updates take
effect without re-registering the webhook.
Stacked PR — base is feat/webhooks-from-orkes-split. Don't merge until
the parent (#1106) lands. After parent merges this rebases onto main.
Added
- V16__webhook.sql: four tables — webhook, incoming_webhook_event,
webhook_target_workflows, webhook_hash_to_taskid. Renamed
webhook_matchers -> webhook_target_workflows to reflect that we
store target workflow ids (override snapshot), NOT pre-computed
matchers. Index on incoming_webhook_event(created_on) for the
cleanup follow-up.
- org.conductoross.conductor.postgres.dao.PostgresWebhookDAO:
9 interface methods + private computeMatchers helper. Extends
PostgresBaseDAO. ON CONFLICT DO UPDATE for upserts.
- org.conductoross.conductor.postgres.dao.PostgresWebhookTaskService:
3 interface methods (put/get/remove). ON CONFLICT DO NOTHING on
put for idempotency.
- PostgresWebhookDAOTest (8 tests with testcontainers + @MockBean MetadataDAO),
PostgresWebhookTaskServiceTest (6 tests with testcontainers).
Critical regression test: getMatchers_recomputesFromMetadataDAO_onEachCall_noStaleCache.
Shared hash extraction
- WebhookTaskHashing helper added to core
(org.conductoross.conductor.service.webhook). Static methods that
both InMemory and Postgres impls use, so tasks registered via one
backing are findable by hash via any other. Validates matches
before composing the hash (regression fix in this commit: NPE in
removeIterationFromTaskRefName would mask the missing-matches
exception).
- InMemoryWebhookTaskService refactored to delegate to the helper
(-50 LoC). Test surface unchanged.
PostgresConfiguration wiring
- Two new @Bean methods with @DependsOn({"flywayForPrimaryDb"}).
Bean names "webhookDAO" and "webhookTaskService" match the
@ConditionalOnMissingBean(name=...) on the InMemory impls, so the
postgres beans take precedence when this module is on the classpath.
Always-on for now (no @ConditionalOnProperty); the whole
postgres-persistence module is gated by conductor.db.type=postgres
at the upstream configuration.
OSS-side design changes vs orkes' PostgresWebhookDAO
- All org_id columns and WHERE clauses dropped. Tables are tenant-free.
- ConductorLimitsDAO/LimitUtils quota check dropped (Orkes-only).
- Preconditions.checkNotNull replaced with NotFoundException where
appropriate; nulls otherwise propagate per OSS convention.
- Scheduled refreshMatchers executor dropped — matcher staleness is
fixed by design (recompute on read).
- workflowDefToUpdateMap cache dropped — same reason.
- The "matches" persistence (webhook_matchers table in orkes)
replaced with "targets" persistence (webhook_target_workflows).
Different shape, same external contract.
Verified locally
- :conductor-postgres-persistence:compileJava + compileTestJava clean
- :conductor-postgres-persistence:spotlessCheck clean
- :conductor-webhooks-oss:test green (81 tests, hashing refactor
doesn't break anything)
- :conductor-server:compileJava clean
Note: postgres tests need Docker daemon for testcontainers; not run
locally. CI will validate.
* fix(postgres-persistence): tolerate orphan migrations in webhook DAO tests
CI 'build' job on #1110 failed with FlywayValidateException:
Detected applied migration not resolved locally: 10.1
V10.1__notify.sql lives in db/migration_postgres_notify/, which
PostgresConfiguration only loads when
conductor.postgres.experimental-queue-notify=true. When another test
class in the same Gradle test executor JVM enables that flag, V10.1
gets applied to the shared testcontainers postgres. When our webhook
tests run without that flag, Flyway validates schema_history against
our (smaller) resolved-migration list and refuses to migrate.
Fix: add spring.flyway.ignore-migration-patterns=*:missing to both
PostgresWebhookDAOTest and PostgresWebhookTaskServiceTest. This tells
Flyway it's OK for the DB to contain applied migrations that aren't in
the current location set — the validation behavior we actually want for
test contexts that share a postgres container across configurations.
The before() @Before hook still flyway.clean()s the DB between tests so
schema state stays predictable within a class.
Local: still passes spotlessCheck; can't run tests (no Docker on dev
box). CI will validate.
* feat(persistence): WebhookDAO + WebhookTaskService impls for SQLite and MySQL
Adds the same in-DB shape we landed for postgres to the two other SQL
backends OSS conductor supports. Cassandra and Redis are deferred to a
follow-up — they need different data modeling (not SQL).
Pattern is identical across backends, dialect specifics aside:
SQLite (sqlite-persistence)
- V4__webhook.sql: 4 tables matching the postgres layout (no org_id)
- SqliteWebhookDAO + SqliteWebhookTaskService extending SqliteBaseDAO
- Uses SQLite's INSERT ... ON CONFLICT (col) DO UPDATE SET col = excluded.col
- Bean wiring in SqliteConfiguration with names webhookDAO + webhookTaskService
MySQL (mysql-persistence)
- V10__webhook.sql: same 4 tables with InnoDB / utf8mb4 + ON UPDATE
CURRENT_TIMESTAMP for modified_on
- MySQLWebhookDAO + MySQLWebhookTaskService extending MySQLBaseDAO
- Uses MySQL's INSERT ... ON DUPLICATE KEY UPDATE col = VALUES(col)
and INSERT IGNORE for the conflict-do-nothing case
- Bean wiring in MySQLConfiguration with the same bean names
Bean name collision is intentional: webhookDAO and webhookTaskService
are the names the InMemory impls' @ConditionalOnMissingBean checks for
in webhooks-oss, so the SQL impls take precedence whenever their
persistence module is on the classpath.
Shared
- All three SQL impls use WebhookTaskHashing from core so the hash a
task is stored under is identical across backings.
- Same matcher-recomputation-on-read design as PostgresWebhookDAO —
target workflows persisted, matchers computed from MetadataDAO on
every getMatchers() call (no stale-cache).
Tests
- SqliteWebhookTaskServiceTest (6) + SqliteWebhookDAOTest (7) — run
against a local SQLite file (no Docker). All 13 pass locally.
Notable per-test workflow names in the task service test to avoid
hash collisions across the shared SQLite file.
- MySQLWebhookTaskServiceTest (6) + MySQLWebhookDAOTest (6) — same
shape; uses jdbc:tc:mysql:8.0.29 testcontainers. CI will validate.
Not run locally — needs Docker.
Known follow-ups (separate PRs)
- Redis adapter (different data model — Redis hashes / sorted sets)
- Cassandra adapter (denormalized CQL)
- WebhookCleanupJob for retention of incoming_webhook_event rows
* feat(redis-persistence): WebhookDAO + WebhookTaskService impls for Redis
Adds the Redis backing. Different paradigm than the SQL impls (no
schema, no SQL) but the same external behavior including matcher
recomputation on read.
Data model
- nsKey("WEBHOOK_CONFIG") hash: webhook_id -> JSON(WebhookConfig)
- nsKey("WEBHOOK_EVENT") hash: event_id -> JSON(IncomingWebhookEvent)
- nsKey("WEBHOOK_TARGETS") hash: webhook_id -> JSON(Map<workflowName,version>)
- nsKey("WEBHOOK_HASH", hash) set: members are task_ids waiting on
that hash
Single hash per concept rather than per-id keys keeps getAllWebhooks()
fast (HVALS) and avoids polluting the Redis keyspace.
OSS pattern alignment
- Both classes extend BaseDynoDAO (gives jedisProxy, nsKey, toJson,
readValue).
- @Component(value="...") matches the bean name the InMemory impls
guard against via @ConditionalOnMissingBean — so Redis wins when its
module is on the classpath and conductor.db.type is set to a Redis
flavor.
- @Conditional(AnyRedisCondition.class) — AnyNestedCondition that
activates for db type memory, redis_cluster, redis_sentinel, or
redis_standalone. Same gate RedisFileMetadataDAO uses.
- WebhookTaskHashing is shared with the SQL impls so the hash a task
is stored under is identical across backings.
Tests
- RedisWebhookTaskServiceTest (6) + RedisWebhookDAOTest (7) — both
use the existing pattern of GenericContainer("redis:7-alpine") +
JedisPool + flushAll() between tests (mirrors RedisMetadataDAOTest).
- @MockBean MetadataDAO in the DAO test to drive the
recomputes-on-read regression.
Local: compiles + spotless clean. Tests not run locally (need Docker).
CI will validate.
* feat(cassandra-persistence): WebhookDAO + WebhookTaskService impls for Cassandra
Final SQL/NoSQL backend for the matching-scheduler-precedent series.
Cassandra-flavored impls of WebhookDAO + WebhookTaskService, following
the org.conductoross.conductor.cassandra.dao.* layout the scheduler
cassandra port uses.
Data model
- webhook (bucket, webhook_id, json_data) with composite PK
((bucket), webhook_id). Single ALL bucket so getAllWebhooks() is a
single-partition scan. Anti-pattern at high cardinality, fine for
admin-scale config lists.
- incoming_webhook_event (event_id PRIMARY KEY, json_data) — high
cardinality, no listing needed.
- webhook_target_workflows (webhook_id PRIMARY KEY, json_data) —
target workflow versions snapshot per webhook. Matchers themselves
are recomputed from MetadataDAO on read (same design as the SQL
backings — no stale cache).
- webhook_hash_to_taskid (hash, task_id) with composite PK
((hash), task_id) — efficient get(hash) via single-partition scan,
efficient point delete by (hash, task_id).
Tables created on construction via ensureTables() — mirrors the
scheduler-cassandra pattern. PreparedStatements built once at
construction, bound per call.
Shared
- WebhookTaskHashing from core — hashes match across backings.
- Local toJson/readValue helpers because the ones in CassandraBaseDAO
are package-private. Same workaround the scheduler-cassandra DAO
applies (and that the file says explicitly: 'objectMapper is private
in CassandraBaseDAO; keep a local reference for serialization').
Tests
- CassandraWebhookDAOTest (7) + CassandraWebhookTaskServiceTest (6)
using testcontainers CassandraContainer('cassandra:3.11.2') +
@ClassRule. Drop tables in @Before for test isolation. Mirrors
CassandraSchedulerDAOTest setup verbatim.
Local: compiles + spotless clean. Tests not run locally (need Docker).
CI will validate.
Matching-scheduler-precedent series complete
All 5 OSS persistence backends now have WebhookDAO + WebhookTaskService
impls: postgres, mysql, sqlite, redis, cassandra. Same external
contract, backing-specific data model.
Still on the deferred list: WebhookCleanupJob for retention of
incoming_webhook_event rows.
* feat(postgres-persistence): PostgresWebhookCleanupJob — scheduled retention for incoming_webhook_event
Periodically deletes old rows from the incoming_webhook_event table so it
doesn't grow unbounded. Postgres-only for now — the other backends will
get their own impls (or rely on Cassandra TTL / Redis per-key expiry) in
follow-up PRs.
Behavior
- @Scheduled cron defaults to hourly ("0 0 * * * *")
- Default retention: 7 days
- Default batch size: 1000 rows per pass
- Default max-runtime: 60s per tick
- Properties: conductor.webhooks.cleanup.{cron,retention-duration,batch-size,max-runtime,enabled}
- enabled defaults to true; users can disable with conductor.webhooks.cleanup.enabled=false
SQL: WITH old_records AS (SELECT ctid WHERE created_on < ? LIMIT ?),
deleted AS (DELETE WHERE ctid IN (...) RETURNING ctid) SELECT count(1)
FROM deleted. CTE + RETURNING keeps each batch's transaction small and
cancellable.
Wired in PostgresConfiguration as a @Bean (not @Component) so it only
loads when conductor.db.type=postgres. The @ConditionalOnProperty for
the enabled flag gates loading further. @EnableScheduling is already
declared in core/SchedulerConfiguration so the @Scheduled annotation
fires once the bean is registered.
Tests
- PostgresWebhookCleanupJobTest (4 tests with testcontainers postgres):
- deletes rows older than retention, keeps recent ones
- empty table is a no-op (no exception)
- all-recent table keeps everything
- batched delete spans multiple passes correctly (batch=2, 7 rows,
forces 4 passes)
Local: compiles + spotless clean. Tests not run locally (need Docker).
CI will validate.
* feat(persistence): MySQL + SQLite WebhookCleanupJob siblings
Completes scheduled retention of incoming_webhook_event rows across all
SQL backings (after PostgresWebhookCleanupJob from the prior commit).
MySQL impl (MySQLWebhookCleanupJob)
- MySQL doesn't support DELETE ... RETURNING the way postgres does, so
uses DELETE FROM ... WHERE created_on < ? LIMIT ? and relies on
PreparedStatement.executeUpdate() return value to know when to stop.
SQLite impl (SqliteWebhookCleanupJob)
- SQLite's DELETE ... LIMIT is only available when compiled with
SQLITE_ENABLE_UPDATE_DELETE_LIMIT (not the default in many distros),
so uses a rowid subquery: DELETE FROM t WHERE rowid IN
(SELECT rowid FROM t WHERE created_on < ? LIMIT ?).
Both follow the postgres impl's contract:
- Default retention: 7 days
- Default cron: hourly
- Default batch size: 1000
- Default max-runtime: 60s per tick
- Properties: conductor.webhooks.cleanup.{cron,retention-duration,
batch-size,max-runtime,enabled}
Wired as @Bean in MySQLConfiguration and SqliteConfiguration with the
same @ConditionalOnProperty(enabled, matchIfMissing=true) gate.
Tests
- MySQLWebhookCleanupJobTest (3) — uses testcontainers MySQL. Not run
locally; CI will validate.
- SqliteWebhookCleanupJobTest (3) — uses local sqlite file; passes
locally (3/3). Required an explicit DELETE FROM in @Before because
flyway.clean() leaks rows across tests on the shared sqlite file.
Cassandra and Redis cleanup are deferred to follow-ups — both have
different idioms (Cassandra TTL on table or per-row, Redis per-key
expiry or hash field walker) that warrant their own design pass.
* feat(persistence): Cassandra TTL + Redis cleanup job — completes retention coverage
All 5 OSS persistence backends now have a retention mechanism for
incoming_webhook_event:
Cassandra (CassandraWebhookDAO)
- Adds default_time_to_live to incoming_webhook_event at CREATE TABLE
time. Cassandra auto-expires rows after the configured window — no
scheduled job needed, no tombstone churn beyond what the SSTable
layer already manages.
- Default TTL: 7 days (matches the SQL cleanup-job default).
- Configurable via conductor.webhooks.cleanup.retention-duration on
the CassandraConfiguration @Bean wiring (Duration → toSeconds for
default_time_to_live).
- Caveat documented inline: CREATE TABLE IF NOT EXISTS won't update an
existing table's TTL; operators changing retention on an existing
deployment need to ALTER TABLE manually.
- The other 3 tables (webhook, webhook_target_workflows,
webhook_hash_to_taskid) don't get TTL — they're durable config /
task-routing state, not auditable events.
Redis (RedisWebhookCleanupJob)
- @Component @Conditional(AnyRedisCondition.class)
@ConditionalOnProperty(conductor.webhooks.cleanup.enabled,
matchIfMissing=true).
- @Scheduled walker on the same cron as the SQL backings: HGETALL on
the WEBHOOK_EVENT hash, parse each value as IncomingWebhookEvent,
HDEL any whose timeStamp is older than retentionDuration.
- Cheap at low cardinality; documented that high-volume deployments
should migrate to per-key TTL storage (HEXPIRE landed in Redis 7.4
but isn't ubiquitous yet).
- 4 tests with testcontainers Redis: deletes old, keeps recent, empty
hash no-op, all-recent kept, timeStamp=0 fixtures preserved.
Verified locally: full compile clean across cassandra/redis modules
and server. Tests not run locally — both need Docker. CI will validate.
Deferred items now all addressed:
- ~~Worker DLQ semantics~~ (in #1106)
- ~~Matcher cache staleness~~ (in #1106 + #1110)
- ~~Postgres impls~~ (#1110)
- ~~MySQL/SQLite/Redis/Cassandra adapters~~ (#1110)
- ~~WebhookCleanupJob (postgres)~~ (#1110)
- ~~MySQL + SQLite cleanup~~ (#1110)
- ~~Cassandra TTL + Redis cleanup~~ (this commit)
Remaining open: EventMessage audit-table persistence (the addEventMessage
→ log.warn debt) — separate, larger architectural decision.
* refactor(webhooks): extract WebhookMatcherComputer to core
Lifts the identical `computeMatchers` body from all 6 WebhookDAO impls
(InMemory + Postgres/MySQL/SQLite/Redis/Cassandra) into a single static
helper in core. Each impl's getMatchers() collapses to a one-liner that
loads the target-workflow snapshot and delegates the WorkflowDef →
matchers transformation.
Behavior identical — the loop body is lifted verbatim. Net –166 LoC.
Same parity-of-behavior argument as WebhookTaskHashing: a WorkflowDef
edit now produces identical matcher output regardless of which
persistence module is backing WebhookDAO, enforced by a shared call
rather than five hand-maintained copies.
* fix(webhook/verifier): stop leaking Stripe signing secret in error response
The 'header is not present' error path concatenated webhookConfig.getSecretValue()
into the response body. Renames the local to signingSecret and uses the literal
STRIPE_SIGNATURE constant in the error message.
* fix(webhook/verifier): guard SignatureBasedVerifier against short header
substring(SHA_256.length()) on a header that doesn't start with 'sha256='
threw an uncaught StringIndexOutOfBoundsException, surfacing as HTTP 500.
Validate the prefix and return a clean ErrorList instead.
* fix(webhook/verifier): use constant-time signature comparison
String.equals on HMAC signatures short-circuits on the first differing byte,
which leaks signature bytes to a network-adjacent attacker with a stopwatch.
Switch to MessageDigest.isEqual in both HMACVerifier and SignatureBasedVerifier.
* fix(webhook): drop signature/payload material from debug logs
HMACVerifier and SignatureBasedVerifier logged both signatures at debug level,
and IncomingWebhookResource logged the full request body, params, and headers map.
With DEBUG enabled in any deployment those become live secret/PII channels — log
only the webhook id and sizes.
* fix(webhook): null-check verifier lookup before invocation
The verifier name comes from a deserialized config; the read path doesn't
re-validate against the live enum. A drift between persisted verifier strings
and registered @Component beans NPE'd. Reject with a clean error instead.
* fix(webhook): drop @SneakyThrows from handlePing
handlePing has no checked-exception surface — verifier.handlePing() declares
none — so the annotation was vestigial obfuscation. Removed it and applied the
same null-verifier guard that handleWebhook now uses for symmetry.
* fix(webhook): validate matches map key types before downstream use
Replaces the @SuppressWarnings(unchecked) cast with a key-type check, copying
into a typed Map<String, Object> when valid and skipping the entry otherwise.
A non-String key now surfaces as a skipped matcher, not a downstream CCE.
* docs(webhook): document MetadataDAO fetch cost on the matcher hot path
The compute() loop issues one getWorkflowDef per target workflow on every read.
Cassandra wraps with CacheableMetadataDAO, but postgres/mysql/sqlite/redis hit
the backend each time — acceptable for small fanout, revisit if matchers fan out.
* docs(webhook/migrations): clarify webhook_hash_to_taskid column intent
The 'hash' column actually holds a delimited deterministic key, not a SHA-256
hash — TEXT in postgres/sqlite is intentional, but MySQL's VARCHAR(255) is at
latent truncation risk on long keys. Documented and flagged for follow-up.
* docs(webhook): document auth model for /api/metadata/webhook endpoints
The webhook config endpoints follow conductor OSS's standard 'no auth in the
server, secure via deployment wrapper' pattern. Added a deployment section
that calls out the two surfaces (public event vs operator metadata) explicitly.
* fix(webhook/verifier): implement Slack's v0= signing-secret protocol
Replaces the urlVerified-only check (which became a no-op after the first
successful event) with the real Slack v0=hex(hmac-sha256(timestamp:body))
verification, plus a 5-min replay-tolerance window. Tests rewritten to
exercise the signed protocol end-to-end.
* feat(webhook/cleanup): cluster-safe cleanup lease across cleanup jobs
Without a lease every replica ran the cleanup cron simultaneously, hammering
the incoming_webhook_event table. SQL backends gain a webhook_cleanup_lease
table acquired via a single conditional UPDATE; Redis uses SET NX PX.
* feat(webhook): signature-dedup replay protection across verifiers + DAOs
Verifiers expose dedupKey() (signature header value); IncomingWebhookService
records it via webhookDAO.tryRecordSignature() after verify() and rejects
duplicates within REPLAY_DEDUP_TTL (5 min). Cassandra retains the default
no-op pending an IF-NOT-EXISTS + TTL impl in a follow-up.
* fix(cassandra/webhook): qualify prepared statements with keyspace
CassandraWebhookDAO and CassandraWebhookTaskService created tables as
<keyspace>.<table> in ensureTables() but their session.prepare() calls
referenced bare table names. With a session not bound to a keyspace via
USE, prepare fails with "No keyspace has been specified" and every test
in CassandraWebhookDAOTest / CassandraWebhookTaskServiceTest aborts at
setUp. Mirror the qualification ensureTables() already does.
* fix(persistence): spring-web on cassandra+redis test classpath
IncomingWebhookEvent.headers is typed as org.springframework.http.HttpHeaders
(mirroring Orkes). Jackson reflects over all declared fields during
serialization, so a NoClassDefFoundError fires whenever a persistence test
serializes the event and spring-web isn't on its test runtime classpath.
postgres/mysql/sqlite already pull spring-web transitively via
:conductor-server testImpl; cassandra and redis only have spring-boot-starter
compileOnly which doesn't include spring-web. Add spring-web as
testImplementation in those two modules.
* style: apply spotless formatting to webhook DAO and matcher computer
* style: apply spotless formatting across webhook modules
* fix(persistence/test): commit inserts in MySQL+Postgres cleanup tests
src/test/resources/application.properties sets
spring.datasource.hikari.auto-commit=false in both modules. The new
MySQLWebhookCleanupJobTest and PostgresWebhookCleanupJobTest insert seed
rows via raw JDBC (dataSource.getConnection() + executeUpdate()) and
let the try-with-resources close the connection without committing —
so the inserts roll back when the connection returns to the pool and
the precondition assertEquals(N, rowCount()) sees 0.
Add an explicit conn.commit() after each insertEvent() executeUpdate.
The cleanup job itself already calls setAutoCommit(true) on its own
connection so its DELETEs are unaffected.
* fix(postgres/test): include notify migration location in webhook tests
The Postgres webhook tests set spring.flyway.ignore-migration-patterns
in @TestPropertySource, but PostgresConfiguration builds its Flyway bean
via Flyway.configure() directly rather than via Spring's auto-config,
so that property has no effect.
Other tests in this module (PostgresQueueListenerTest,
PostgresGrpcEndToEndTest) set experimentalQueueNotify=true and apply
V10.1 from migration_postgres_notify to the shared testcontainer DB.
When the new webhook tests run after them in the same JVM, their
Flyway has only migration_postgres on its locations, sees V10.1 in
flyway_schema_history, and aborts context creation with "Detected
applied migration not resolved locally: 10.1" — taking all 18 tests
in the three webhook test classes down with it.
Switch the webhook tests to also set experimentalQueueNotify=true so
their Flyway includes the notify location and validation passes
regardless of test ordering. The notify triggers only fire on
queue_message writes, which these tests never do, so there's no
behavioral impact.
* fix(postgres/test): truncate tables instead of flyway.clean()
PostgresConfiguration builds its Flyway bean via Flyway.configure()
directly rather than via Spring's auto-config, so neither
spring.flyway.clean-disabled=false nor any other spring.flyway.* test
property applies to it. Flyway defaults to cleanDisabled=true, so the
@Before's flyway.clean() throws FlywayException across all 18 webhook
tests.
Drop flyway.clean()/migrate() and truncate the relevant tables
directly in each test class:
- PostgresWebhookCleanupJobTest: truncate incoming_webhook_event +
reset the webhook_cleanup_lease row (tryAcquireLease updates but
never releases — second test in this class needs the lease back).
- PostgresWebhookDAOTest: truncate webhook, incoming_webhook_event,
webhook_target_workflows.
- PostgresWebhookTaskServiceTest: truncate webhook_hash_to_taskid.
PostgresQueueListenerTest in this module uses the same TRUNCATE pattern
for the same reason.
* fix(sqlite/test): reset cleanup lease row in @Before
SqliteWebhookCleanupJobTest already worked around flyway.clean()'s
incomplete reset on the shared sqlite file by adding an explicit
DELETE FROM incoming_webhook_event in @Before, but missed the
webhook_cleanup_lease row. V5__webhook_cleanup_lease.sql seeds the
row with INSERT OR IGNORE, so re-running flyway.migrate() after a
partial-clean is a no-op and the prior test's expires_at = now + 5min
sticks. The first test acquires the lease and the next two tests'
tryAcquireLease() returns false, so job.run() exits early and
nothing is deleted — failing the post-cleanup row-count assertions.
Add an explicit UPDATE of the lease row alongside the existing
incoming_webhook_event delete so every test starts with the lease
available.
* test(sendgrid): add SendGridVerifierTest — missing headers, bad key, sig pass/fail
* test(incoming-webhook): add IncomingWebhookServiceTest — all handleWebhook + handlePing branches
* test(webhook-worker): add non-Map body and recordHistory trim-when-full tests
* test(webhook-hashing): add WebhookHashingServiceTest — non-Map body, empty array, malformed JSON
* refactor(test): replace mocks with real in-memory impls — IncomingWebhookServiceTest (0 mocks), WebhookWorkerTest (2 mocks from 5)
* fix(sqlite/test): reset cleanup lease with parameterized Timestamp
The previous attempt set webhook_cleanup_lease.expires_at via a text
literal ('1970-01-01T00:00:00'), but the xerial sqlite-jdbc driver
stores java.sql.Timestamp as INTEGER (epoch millis) and SQLite's type
comparison rule says INTEGER < TEXT — so the cleanup job's
"expires_at < ?" check (with ? bound as a Timestamp/INTEGER) compares
TEXT > INTEGER → false, and the lease could never be re-acquired.
Switch to INSERT OR REPLACE with the expires_at value bound via
ps.setTimestamp() so it lands as INTEGER in the same format the
cleanup job's comparison uses.
Note: V5__webhook_cleanup_lease.sql seeds the row with the same broken
text literal, so a single-instance SQLite deployment running the
cleanup job for the first time would also fail to acquire its lease.
Follow-up needed on the V5 migration itself, but out of scope here.
* fix(sqlite/webhook): seed cleanup lease expires_at as INTEGER
Closes #1142
V5 seeded webhook_cleanup_lease.expires_at with a text literal
('1970-01-01T00:00:00'). The xerial sqlite-jdbc driver stores
java.sql.Timestamp as INTEGER (epoch millis), and SQLite's type-
comparison rule says INTEGER < TEXT — so SqliteWebhookCleanupJob's
"expires_at < ?" check (with ? bound as a Timestamp) was always
false on a freshly migrated DB. The cleanup job would log "lease
held elsewhere" forever and never delete an incoming_webhook_event
row.
V5 hasn't shipped to main yet, so fix the seed value in place
rather than adding a patch migration. Use INTEGER 0 so the seed's
storage class matches what the cleanup job writes via setTimestamp().
The test's @Before lease reset (added earlier in this PR for between-
test isolation, since flyway.clean() leaks rows on the shared sqlite
file) remains necessary — it covers a different concern. Trim its
comments to drop the now-stale "V5 seed is broken" wording.
Postgres and MySQL aren't affected: their native TIMESTAMP types parse
the text literal as a real timestamp. No Orkes parallel — sqlite-
persistence and the cleanup-lease pattern are both OSS-side additions.
* style: apply spotless to webhooks-oss test files
* remove Tag.java from common — tags are enterprise RBAC, no OSS backing
Tag.java was introduced by the webhooks-oss port but serves no purpose in OSS:
tags in Orkes power TagsService/AutomaticTagService-based access control, neither
of which exists here. The class had no callers other than WebhookConfig.tags.
* drop tags field from WebhookConfig — never populated in OSS
In Orkes, WebhookConfig.tags is populated at query-time by TagsService from
a separate enterprise tags table. OSS has no TagsService, so the field was
always null. Dead API surface that would mislead users.
* remove EventMessage.java — ported but never wired, no OSS backing
EventMessage was added to common to mirror Orkes' webhook event audit trail,
but OSS has no ExecutionDAO.addEventMessage() and IncomingWebhookService
never references it. Zero callers anywhere in OSS — pure dead code.
The event audit feature (DLQ persistence for rejected/unmatched webhook events)
requires a separate architectural decision before it belongs here.
* remove WebhookExecutionHistory — enterprise-only, both sides marked TODO Remove
WebhookExecutionHistory embeds a rolling execution log inside WebhookConfig
and is only implemented in orkes-conductor/webhooks-enterprise. Both the Orkes
and OSS models carried a '// TODO Remove this' comment. The recordHistory()
call also held the only webhookDAO.createWebhook() write-back after dispatch,
but urlVerified is already persisted by IncomingWebhookService before the event
is enqueued, so that write was redundant. Drops lastRunWorkflowIdSize property
and its associated test.
* drop WebhookConfig.getWorkflowNames() — zero callers, duplicates map getter
Method was @JsonIgnore and only extracted keys from receiverWorkflowNamesToVersions,
which is already accessible directly. No call site exists anywhere in OSS.
* map NonTransientException → 400 in ApplicationExceptionMapper
NonTransientException previously fell through to the default 500, which
told webhook senders (Stripe, GitHub, Slack) that the server was broken
and to retry. Signature verification failures are client errors — the
sender's signature is wrong — and must return 4xx so senders stop
retrying.
NonTransientException by semantics ("this won't work no matter how many
times you retry") maps cleanly to 400 Bad Request.
Verified: bad_signature and replay cases in negative_smoke.py now return
400 instead of 500, 5/5 negative cases passing on enki.
* fix(webhook): SREM correlation set on task cancel to stop orphan-set growth
Ports orkes-io/orkes-conductor#3663. Webhook.start() does a put() into the
WAIT_FOR_WEBHOOK hash set; the only removal path was WebhookWorker.handleEvent
firing SREM on a matching event. Workflows terminated before a match arrived
left their taskId in the set indefinitely — Redis memory leak / Postgres row
accumulation in webhook_hash_to_taskid.
Fix: add Webhook.cancel() that calls webhookTaskService.remove(TaskModel, int)
on all six backings (postgres, mysql, redis, cassandra, sqlite, in-memory).
WebhookTaskHashing.computeHashIfPresent() handles tasks cancelled before
IN_PROGRESS (null/missing matches) as a safe no-op instead of throwing.
Cleanup failures are caught and logged so the terminate path is never blocked.
---------
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
||
|
Kowser
|
795566f111 | fix: Bring back SqliteFileMetadataDAO bean | ||
|
Manan Bhatt
|
ce5b05ad60 |
fix(redis): release postponed task when concurrencyLimit slot frees (#1105)
* fix(redis): release postponed task when concurrencyLimit slot frees When a task definition has concurrentExecLimit set and the limit is reached, additional tasks of that type are postponed by a static queueTaskMessagePostponeSecs. If an in-progress task completes and frees a slot before that duration elapses, the postponed task continues to wait until the postpone window expires. On terminal status in RedisExecutionDAO.updateTask, peek the next pending message in that queue via zrangeByScore and call queueDAO.resetOffsetTime to set its score to now. The next worker poll picks it up immediately. * fix(postgres): release postponed task when concurrencyLimit slot frees Extends the redis-side fix to the postgres backend. Adds peekFirstIds to PostgresQueueDAO (SELECT non-popped message_ids by deliver_on / priority / created_on). In PostgresExecutionDAO.updateTask, on terminal status of a concurrencyLimit-bound task, peek the next pending message and call queueDAO.resetOffsetTime so the next worker poll picks it up immediately instead of waiting out the static postpone window. Injects QueueDAO into PostgresExecutionDAO; bean wiring updated. * style: apply spotless formatting * test(e2e): add concurrency-slot wakeup repro to ConcurrentExecLimitTests Adds testPostponedTaskReleasedOnSlotFree: with concurrentExecLimit=1 and a FORK_JOIN of two tasks of that type, poll once to occupy the slot (task A), poll again to put task B in the postponed state, complete A, then assert B becomes pollable within 5s. Pre-fix (without #1105): B waits the full taskExecutionPostponeDuration (~60s default). Post-fix: <100ms because terminal completion of A peeks the queue and resets B's offset_time to now. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(sqlite,mysql,cassandra): release postponed task when concurrencyLimit slot frees Extends the slot-release wakeup added for Redis and Postgres in this PR to the remaining ExecutionDAO implementations so SQLite, MySQL, and Cassandra users get the same fix. SQLite and MySQL: - QueueDAO override of peekFirstIds: SELECT message_id FROM queue_message WHERE queue_name=? AND popped=false ORDER BY deliver_on, priority DESC, created_on LIMIT ? - ExecutionDAO.updateTask now peeks the queue and calls resetOffsetTime when a concurrentExecLimit task hits a terminal status. - Configuration wires the local QueueDAO into the ExecutionDAO bean. Cassandra: - CassandraExecutionDAO.updateTask calls queueDAO.peekFirstIds + queueDAO.resetOffsetTime alongside the existing removeTaskFromLimit branch when a concurrentExecLimit task is terminal. - There is no CassandraQueueDAO; Cassandra users pair it with another queue impl. Since the QueueDAO interface defaults peekFirstIds to an empty list, this is a no-op when paired with a queue that hasn't overridden peekFirstIds. |
||
|
kowser-orkes
|
09aa35e196 | Initial implementation for file storage feature | ||
|
Viren Baraiya
|
6456b564fa | Improve task retry policy (#1031) | ||
|
Viren Baraiya
|
fec4f90e20 | feat: wildcard search and sub-workflow filtering (#996) | ||
|
Viren Baraiya
|
30fc6adc73 |
fix(persistence): strip single quotes from search query values (#950)
* fix(persistence): strip single quotes from search query values The query value parser in both SqliteIndexQueryBuilder and PostgresIndexQueryBuilder only stripped double quotes and parentheses from condition values, but not single quotes. This caused searches like workflowId='some-id' to match against the literal value 'some-id' (with embedded quotes) instead of some-id, returning zero results. This also broke the internal getWorkflowsByCorrelationId code path in ExecutionDAOFacade which constructs queries using single quotes. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * test(e2e): add workflow search e2e tests for single-quote queries Adapted from orkes-conductor WorkflowSearchTests. Adds end-to-end tests that start a real Conductor server and exercise the full search flow: - testSearchByWorkflowIdWithDoubleQuotes: baseline workflowId search - testSearchByWorkflowIdWithSingleQuotes: validates the single-quote fix - testSearchV2ByWorkflowId: searchV2 (returns full Workflow objects) - testSearchByWorkflowTypeWithSingleQuotes: workflowType with single quotes - testSearchWithMultipleConditionsAndSingleQuotes: multi-condition query mimicking the getWorkflowsByCorrelationId internal code path - testSearchWithPaginationAndSort: pagination and sort with single quotes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * test(e2e): remove searchV2 test (pre-existing server bug) PostgresIndexDAO.searchWorkflows returns null (unimplemented), causing NPE in ExecutionService.searchV2. This is a pre-existing issue unrelated to the single-quote fix. All remaining 6 e2e tests pass against a real Conductor server with Postgres persistence. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(core): make IDGenerator a resilient fallback bean IDGenerator used @ConditionalOnProperty(havingValue="default", matchIfMissing=true), which meant setting conductor.id.generator to any value other than "default" (even a typo or non-existing implementation) would prevent the bean from being created, crashing the server at startup. Changed to @ConditionalOnMissingBean on a @Bean method in ConductorCoreConfiguration. This means the default IDGenerator is always available unless a custom implementation bean is explicitly provided, regardless of property values. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix formatting * Update PostgresIndexDAOTest.java --------- Co-authored-by: AgentSpan Coder <coder@agentspan.ai> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com> |
||
|
Viren Baraiya
|
4cb030bf08 | Refactor redis and upgrade jedis (#927) | ||
|
Viren Baraiya
|
adb29cfbdb |
Jdbc workers (#878)
* JDBC workers * Documentation update |
||
|
Viren Baraiya
|
775634f1da | Fix issue with sqlite based queue where the timeout did not work consistently at sub-second level (#738) | ||
|
Viren Baraiya
|
368f7abc31 | Update SqliteIndexDAOTest.java | ||
|
Viren Baraiya
|
99ecc5f126 | sweeper fixes and sync execution | ||
|
Viren Baraiya
|
ba24100035 | remove legacy clients | ||
|
Orkes
|
f82805aff2 | removing print statment | ||
|
Orkes
|
bcb62681f8 | fixing properties and configuration | ||
|
Orkes
|
e3143c0e5a | fixing_testcases | ||
|
Orkes
|
a530fd7ef6 | Feature: Fixing Index and QueueDAO and final fixes | ||
|
Orkes
|
edd0d4c9a3 | spotlessApply | ||
|
Orkes
|
de617c1904 | Adding PollData DAO | ||
|
Orkes
|
67318c8da2 | Feature: Implementing ExecutionDAO for Sqlite | ||
|
Orkes
|
0ba916ab3d | implementing queuedao in sqlite | ||
|
Orkes
|
fe5d09b175 | fixing-testcase and remvoing component | ||
|
Orkes
|
3160404073 | applying spot less check | ||
|
Orkes
|
f369aee111 | Merge branch 'sqlite_persistence' into sqlite_persistence_mtdao | ||
|
Orkes
|
7ef1d92fe1 | deleting backfill_update | ||
|
Orkes
|
266ed4c39e | Feature: sqlite implementation for metadataDAO | ||
|
Orkes
|
d8a8f00770 | applying spotless checks | ||
|
Orkes
|
005aa0e143 | updating schema | ||
|
Orkes
|
fce17118c9 | adding-sqlite-persistence |