From 5d4d9eb50da087a796bcf980e3fca9972fe1efee Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Wed, 3 Jun 2026 23:09:45 -0300 Subject: [PATCH] feat(jobs): add job_events table and JobEvent model with unique (job_id, seq) --- .../b026885b89c8_add_job_events_table.py | 51 +++++++++++++++++++ .../services/database/models/__init__.py | 3 +- .../services/database/models/jobs/__init__.py | 4 +- .../services/database/models/jobs/model.py | 29 ++++++++++- .../test_background_execution_migrations.py | 19 +++++++ 5 files changed, 101 insertions(+), 5 deletions(-) create mode 100644 src/backend/base/langflow/alembic/versions/b026885b89c8_add_job_events_table.py diff --git a/src/backend/base/langflow/alembic/versions/b026885b89c8_add_job_events_table.py b/src/backend/base/langflow/alembic/versions/b026885b89c8_add_job_events_table.py new file mode 100644 index 0000000000..7c9dddd7fd --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/b026885b89c8_add_job_events_table.py @@ -0,0 +1,51 @@ +"""add job_events table for durable background-job event log. + +Revision ID: b026885b89c8 +Revises: 185482a2d715 +Create Date: 2026-06-03 10:05:00.000000 + +Phase: EXPAND +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "b026885b89c8" # pragma: allowlist secret +down_revision: str | None = "185482a2d715" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql") + + +def upgrade() -> None: + from langflow.utils import migration + + conn = op.get_bind() + if not migration.table_exists("job_events", conn): + op.create_table( + "job_events", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("job_id", sa.Uuid(), nullable=False), + sa.Column("seq", sa.Integer(), nullable=False), + sa.Column("event_type", sa.String(), nullable=False), + sa.Column("payload", _JSON, nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("job_id", "seq", name="uq_job_events_job_id_seq"), + ) + with op.batch_alter_table("job_events", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_job_events_id"), ["id"], unique=False) + batch_op.create_index(batch_op.f("ix_job_events_job_id"), ["job_id"], unique=False) + + +def downgrade() -> None: + from langflow.utils import migration + + conn = op.get_bind() + if migration.table_exists("job_events", conn): + op.drop_table("job_events") diff --git a/src/backend/base/langflow/services/database/models/__init__.py b/src/backend/base/langflow/services/database/models/__init__.py index 204c91b7de..63f0f5c875 100644 --- a/src/backend/base/langflow/services/database/models/__init__.py +++ b/src/backend/base/langflow/services/database/models/__init__.py @@ -19,7 +19,7 @@ from .flow_version import FlowVersion from .flow_version_deployment_attachment import FlowVersionDeploymentAttachment from .folder import Folder from .ingestion_run import IngestionRun, IngestionRunStatus -from .jobs import Job +from .jobs import Job, JobEvent from .knowledge_base import KnowledgeBaseRecord, KnowledgeBaseStatus from .memory_base import MemoryBase, MemoryBaseSession, MemoryBaseWorkflowRun, MessageIngestionRecord from .message import MessageTable @@ -48,6 +48,7 @@ __all__ = [ "IngestionRun", "IngestionRunStatus", "Job", + "JobEvent", "KnowledgeBaseRecord", "KnowledgeBaseStatus", "MemoryBase", diff --git a/src/backend/base/langflow/services/database/models/jobs/__init__.py b/src/backend/base/langflow/services/database/models/jobs/__init__.py index 7330055617..546eea0fe9 100644 --- a/src/backend/base/langflow/services/database/models/jobs/__init__.py +++ b/src/backend/base/langflow/services/database/models/jobs/__init__.py @@ -1,3 +1,3 @@ -from .model import Job +from .model import Job, JobEvent -__all__ = ["Job"] +__all__ = ["Job", "JobEvent"] diff --git a/src/backend/base/langflow/services/database/models/jobs/model.py b/src/backend/base/langflow/services/database/models/jobs/model.py index 86ce5332f1..72d690a549 100644 --- a/src/backend/base/langflow/services/database/models/jobs/model.py +++ b/src/backend/base/langflow/services/database/models/jobs/model.py @@ -1,9 +1,9 @@ from datetime import datetime, timezone from enum import Enum from typing import Any -from uuid import UUID +from uuid import UUID, uuid4 -from sqlalchemy import JSON, Column, DateTime +from sqlalchemy import JSON, Column, DateTime, Integer, String, UniqueConstraint from sqlalchemy import Enum as SQLEnum from sqlalchemy.dialects.postgresql import JSONB from sqlmodel import Field, SQLModel @@ -101,3 +101,28 @@ class JobBase(SQLModel): class Job(JobBase, table=True): # type: ignore[call-arg] __tablename__ = "job" + + +class JobEvent(SQLModel, table=True): # type: ignore[call-arg] + """Durable event log for a background job. + + ``seq`` is the per-job monotonic cursor used as the SSE Last-Event-ID. + UNIQUE(job_id, seq) enforces gap-free ordering and lets append_event + detect collisions. + """ + + __tablename__ = "job_events" + __table_args__ = (UniqueConstraint("job_id", "seq", name="uq_job_events_job_id_seq"),) + + id: UUID = Field(default_factory=uuid4, primary_key=True, index=True) + job_id: UUID = Field(index=True, nullable=False) + seq: int = Field(sa_column=Column(Integer, nullable=False)) + event_type: str = Field(sa_column=Column(String, nullable=False)) + payload: dict[str, Any] | None = Field( + default=None, + sa_column=Column(JsonVariant, nullable=True), + ) + created_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc), + sa_column=Column(DateTime(timezone=True), nullable=False), + ) diff --git a/src/backend/tests/unit/alembic/test_background_execution_migrations.py b/src/backend/tests/unit/alembic/test_background_execution_migrations.py index 03ab8192dc..b2b4ea09a0 100644 --- a/src/backend/tests/unit/alembic/test_background_execution_migrations.py +++ b/src/backend/tests/unit/alembic/test_background_execution_migrations.py @@ -25,3 +25,22 @@ def test_job_has_result_and_error_columns(db_url): # noqa: F811 assert "result" in columns assert "error" in columns + + +def test_job_events_table_and_unique_seq(db_url): # noqa: F811 + alembic_cfg = _make_alembic_cfg(db_url) + command.upgrade(alembic_cfg, "head") + + engine = create_engine(_engine_url(db_url)) + try: + with engine.connect() as connection: + inspector = inspect(connection) + assert "job_events" in inspector.get_table_names() + columns = {c["name"] for c in inspector.get_columns("job_events")} + assert {"id", "job_id", "seq", "event_type", "payload", "created_at"} <= columns + indexes = inspector.get_indexes("job_events") + assert any(idx["column_names"] == ["job_id"] for idx in indexes) + unique = inspector.get_unique_constraints("job_events") + assert any(set(u["column_names"]) == {"job_id", "seq"} for u in unique) + finally: + engine.dispose()