Refactor: Knowledge compilation: move knowledge_compile_docs column backlog_doc_ids and inflight_doc_ids to seperated tables (#21381)

internal/dao/database.go

@@ -206,6 +206,8 @@ func InitDB(ctx context.Context, migrateDB bool) error {
 		&entity.FileCommit{},
 		&entity.FileCommitItem{},
 		&entity.KnowledgeCompileDataset{},
+		&entity.KnowledgeCompileBacklog{},
+		&entity.KnowledgeCompileInflight{},
 		&entity.WikiDocumentDirty{},
 		// Knowledge-compile compilation templates and their groups. The Go
 		// KnowledgeCompilerComponent resolves a compilation_template (or group)
@@ -237,6 +239,9 @@ func InitDB(ctx context.Context, migrateDB bool) error {
 				return fmt.Errorf("failed to migrate model %T: %w", m, err)
 			}
 		}
+		if err = migrateKnowledgeCompileQueues(ctx, DB); err != nil {
+			return fmt.Errorf("failed to split knowledge compilation queues: %w", err)
+		}
 
 		// Split the conversation message and reference payloads out of their
 		// parent tables. It has to run after AutoMigrate, which unlike
@@ -261,6 +266,11 @@ func InitDB(ctx context.Context, migrateDB bool) error {
 		}
 		common.Info("Database schema migrated successfully")
 	} else {
+		for _, column := range []string{"backlog_doc_ids", "inflight_doc_ids"} {
+			if DB.WithContext(ctx).Migrator().HasColumn(&entity.KnowledgeCompileDataset{}, column) {
+				return fmt.Errorf("knowledge compilation queues need migration: stop queue producers and ingestors, then run ragflow_server --migrate")
+			}
+		}
 		if err = migrateIngestionLogRunIdentity(ctx, DB); err != nil {
 			return err
 		}
@@ -361,6 +371,8 @@ func autoMigrateRuntimeModels(ctx context.Context, db *gorm.DB) error {
 		&entity.IngestionTask{},
 		&entity.IngestionTaskLog{},
 		&entity.MemoryTask{},
+		&entity.KnowledgeCompileBacklog{},
+		&entity.KnowledgeCompileInflight{},
 		&entity.ConversationMessage{},
 		&entity.ConversationReference{},
 		&entity.API4ConversationMessage{},

internal/dao/knowledge_compile_queue_migration.go

@@ -0,0 +1,112 @@
+package dao
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"strings"
+
+	"ragflow/internal/entity"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+// migrateKnowledgeCompileQueues runs with queue producers and workers stopped.
+// Each dataset is copied transactionally; rerunning replaces its copied rows
+// until the source column is dropped, so interruption cannot duplicate events.
+func migrateKnowledgeCompileQueues(ctx context.Context, db *gorm.DB) error {
+	db = db.WithContext(ctx)
+	if !db.Migrator().HasTable(&entity.KnowledgeCompileDataset{}) {
+		return nil
+	}
+	for _, queue := range []struct {
+		column string
+		table  string
+	}{
+		{"backlog_doc_ids", (entity.KnowledgeCompileBacklog{}).TableName()},
+		{"inflight_doc_ids", (entity.KnowledgeCompileInflight{}).TableName()},
+	} {
+		if !db.Migrator().HasColumn(&entity.KnowledgeCompileDataset{}, queue.column) {
+			continue
+		}
+		lastDatasetID := ""
+		firstBatch := true
+		for {
+			var datasetIDs []string
+			query := db.Model(&entity.KnowledgeCompileDataset{}).Order("dataset_id ASC").Limit(100)
+			if !firstBatch {
+				query = query.Where("dataset_id > ?", lastDatasetID)
+			}
+			if err := query.Pluck("dataset_id", &datasetIDs).Error; err != nil {
+				return err
+			}
+			if len(datasetIDs) == 0 {
+				break
+			}
+			for _, datasetID := range datasetIDs {
+				if err := db.Transaction(func(tx *gorm.DB) error {
+					var source struct {
+						Payload *string `gorm:"column:payload"`
+					}
+					if err := tx.Model(&entity.KnowledgeCompileDataset{}).
+						Clauses(clause.Locking{Strength: "UPDATE"}).
+						Select(queue.column+" AS payload").Where("dataset_id = ?", datasetID).Take(&source).Error; err != nil {
+						return err
+					}
+					if err := tx.Table(queue.table).Where("dataset_id = ?", datasetID).
+						Delete(&entity.KnowledgeCompileEntry{}).Error; err != nil {
+						return err
+					}
+					if source.Payload == nil || strings.TrimSpace(*source.Payload) == "" || strings.TrimSpace(*source.Payload) == "null" {
+						return nil
+					}
+					decoder := json.NewDecoder(strings.NewReader(*source.Payload))
+					token, err := decoder.Token()
+					if err != nil {
+						return err
+					}
+					if token != json.Delim('[') {
+						return fmt.Errorf("expected an event array")
+					}
+					var position int64
+					for decoder.More() {
+						entries := make([]entity.KnowledgeCompileEntry, 0, 100)
+						for len(entries) < 100 && decoder.More() {
+							var entry entity.KnowledgeCompileEntry
+							if err := decoder.Decode(&entry); err != nil {
+								return err
+							}
+							position++
+							entry.DatasetID = datasetID
+							entry.QueuePosition = position
+							entries = append(entries, entry)
+						}
+						if err := tx.Table(queue.table).Create(&entries).Error; err != nil {
+							return err
+						}
+					}
+					if _, err := decoder.Token(); err != nil {
+						return err
+					}
+					if err := decoder.Decode(new(json.RawMessage)); err != io.EOF {
+						return fmt.Errorf("unexpected trailing data after event array")
+					}
+					return nil
+				}); err != nil {
+					return fmt.Errorf("split knowledge_compile_docs.%s for dataset %s: %w", queue.column, datasetID, err)
+				}
+			}
+			lastDatasetID = datasetIDs[len(datasetIDs)-1]
+			firstBatch = false
+		}
+		if err := db.Migrator().DropColumn(&entity.KnowledgeCompileDataset{}, queue.column); err != nil {
+			return fmt.Errorf("drop knowledge_compile_docs.%s: %w", queue.column, err)
+		}
+		if db.Migrator().HasColumn(&entity.KnowledgeCompileDataset{}, queue.column) {
+			return fmt.Errorf("knowledge_compile_docs.%s still exists after migration", queue.column)
+		}
+	}
+	return nil
+}

internal/dao/knowledge_compile_queue_migration_test.go

@@ -0,0 +1,154 @@
+package dao
+
+import (
+	"encoding/json"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"ragflow/internal/entity"
+
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+)
+
+func setupKnowledgeCompileMigrationDB(t *testing.T) *gorm.DB {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pool, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { pool.Close() })
+	if err := db.AutoMigrate(&entity.KnowledgeCompileDataset{}, &entity.KnowledgeCompileBacklog{}, &entity.KnowledgeCompileInflight{}); err != nil {
+		t.Fatal(err)
+	}
+	for _, column := range []string{"backlog_doc_ids", "inflight_doc_ids"} {
+		if err := db.Exec("ALTER TABLE knowledge_compile_docs ADD COLUMN `" + column + "` LONGTEXT").Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	return db
+}
+
+func TestMigrateKnowledgeCompileQueuesPreservesEventsAndLease(t *testing.T) {
+	db := setupKnowledgeCompileMigrationDB(t)
+	expires := time.Now().Add(time.Minute).UTC()
+	row := entity.KnowledgeCompileDataset{
+		DatasetID: "kb1", TenantID: "tenant1", ClaimOwner: "worker", ClaimToken: "token",
+		ClaimExpiresAt: &expires, State: entity.DatasetStateRunning, Progress: 0.5,
+	}
+	if err := db.Create(&row).Error; err != nil {
+		t.Fatal(err)
+	}
+	entries := make([]entity.KnowledgeCompileEntry, 205)
+	for i := range entries {
+		entries[i] = entity.KnowledgeCompileEntry{DocID: "same-doc", EventType: "doc_completed", Variants: []string{"structure"}, TaskTypes: []string{"Graph"}}
+	}
+	entries[204].EventType = "doc_deleted"
+	payload, err := json.Marshal(entries)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Model(&row).Updates(map[string]interface{}{
+		"backlog_doc_ids": string(payload), "inflight_doc_ids": `[{"doc_id":"d2","event_type":"doc_disabled"}]`,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := migrateKnowledgeCompileQueues(t.Context(), db); err != nil {
+		t.Fatal(err)
+	}
+	for _, column := range []string{"backlog_doc_ids", "inflight_doc_ids"} {
+		if db.Migrator().HasColumn(&entity.KnowledgeCompileDataset{}, column) {
+			t.Fatalf("source column %s was not removed", column)
+		}
+	}
+	var backlog []entity.KnowledgeCompileEntry
+	if err := db.Table((entity.KnowledgeCompileBacklog{}).TableName()).Order("queue_position").Find(&backlog).Error; err != nil {
+		t.Fatal(err)
+	}
+	if len(backlog) != len(entries) {
+		t.Fatalf("backlog count = %d, want %d", len(backlog), len(entries))
+	}
+	for i, got := range backlog {
+		want := entries[i]
+		want.DatasetID = "kb1"
+		want.QueuePosition = int64(i + 1)
+		if !reflect.DeepEqual(got, want) {
+			t.Fatalf("entry %d = %+v, want %+v", i, got, want)
+		}
+	}
+	var inflight entity.KnowledgeCompileInflight
+	if err := db.Take(&inflight).Error; err != nil {
+		t.Fatal(err)
+	}
+	if inflight.DocID != "d2" || inflight.EventType != "doc_disabled" || inflight.Variants != nil || inflight.TaskTypes != nil {
+		t.Fatalf("unexpected inflight entry: %+v", inflight)
+	}
+	var stored entity.KnowledgeCompileDataset
+	if err := db.Where("dataset_id = ?", "kb1").Take(&stored).Error; err != nil {
+		t.Fatal(err)
+	}
+	if stored.ClaimToken != row.ClaimToken || stored.ClaimOwner != row.ClaimOwner || !stored.ClaimExpiresAt.Equal(expires) || stored.State != row.State || stored.Progress != row.Progress {
+		t.Fatalf("migration changed the scheduling state: %+v", stored)
+	}
+	if err := migrateKnowledgeCompileQueues(t.Context(), db); err != nil {
+		t.Fatalf("rerun: %v", err)
+	}
+	var count int64
+	if err := db.Model(&entity.KnowledgeCompileBacklog{}).Count(&count).Error; err != nil || count != 205 {
+		t.Fatalf("rerun duplicated entries: count=%d err=%v", count, err)
+	}
+}
+
+func TestMigrateKnowledgeCompileQueuesResumesAfterInvalidPayload(t *testing.T) {
+	for _, queue := range []struct {
+		column string
+		table  string
+	}{
+		{"backlog_doc_ids", (entity.KnowledgeCompileBacklog{}).TableName()},
+		{"inflight_doc_ids", (entity.KnowledgeCompileInflight{}).TableName()},
+	} {
+		t.Run(queue.column, func(t *testing.T) {
+			db := setupKnowledgeCompileMigrationDB(t)
+			for _, id := range []string{"", "a", "b"} {
+				if err := db.Create(&entity.KnowledgeCompileDataset{DatasetID: id}).Error; err != nil {
+					t.Fatal(err)
+				}
+			}
+			valid := `[{"doc_id":"same","event_type":"doc_completed"},{"doc_id":"same","event_type":"doc_completed"}]`
+			invalid := strings.TrimSuffix(valid, "]")
+			if err := db.Exec("UPDATE knowledge_compile_docs SET backlog_doc_ids = ?, inflight_doc_ids = ?", valid, valid).Error; err != nil {
+				t.Fatal(err)
+			}
+			if err := db.Model(&entity.KnowledgeCompileDataset{}).Where("dataset_id = ?", "b").Update(queue.column, invalid).Error; err != nil {
+				t.Fatal(err)
+			}
+			if err := migrateKnowledgeCompileQueues(t.Context(), db); err == nil {
+				t.Fatal("invalid source must fail migration")
+			}
+			if !db.Migrator().HasColumn(&entity.KnowledgeCompileDataset{}, queue.column) {
+				t.Fatal("failed migration dropped source column")
+			}
+			var count int64
+			if err := db.Table(queue.table).Where("dataset_id = ?", "b").Count(&count).Error; err != nil || count != 0 {
+				t.Fatalf("failed dataset was not rolled back: count=%d err=%v", count, err)
+			}
+			if err := db.Model(&entity.KnowledgeCompileDataset{}).Where("dataset_id = ?", "b").Update(queue.column, valid).Error; err != nil {
+				t.Fatal(err)
+			}
+			if err := migrateKnowledgeCompileQueues(t.Context(), db); err != nil {
+				t.Fatalf("resume: %v", err)
+			}
+			for _, table := range []string{(entity.KnowledgeCompileBacklog{}).TableName(), (entity.KnowledgeCompileInflight{}).TableName()} {
+				if err := db.Table(table).Count(&count).Error; err != nil || count != 6 {
+					t.Fatalf("resumed migration lost or duplicated events in %s: count=%d err=%v", table, count, err)
+				}
+			}
+		})
+	}
+}

internal/dao/migration.go

@@ -644,17 +644,6 @@ func modifyColumnTypes(ctx context.Context, db *gorm.DB) error {
 			columnType: "longtext", nullable: true,
 			definition: "longtext NULL",
 		},
-		// Knowledge-compile queues can exceed MySQL TEXT's 64 KiB limit.
-		{
-			table: "knowledge_compile_docs", column: "backlog_doc_ids",
-			columnType: "longtext", nullable: false,
-			definition: "longtext NOT NULL",
-		},
-		{
-			table: "knowledge_compile_docs", column: "inflight_doc_ids",
-			columnType: "longtext", nullable: false,
-			definition: "longtext NOT NULL",
-		},
 		// system_settings.value mirrors Python's EmptyStringTextField(null=False).
 		{
 			table: "system_settings", column: "value",
@@ -680,18 +669,6 @@ func modifyColumnTypes(ctx context.Context, db *gorm.DB) error {
 			return err
 		}
 	}
-	if db.WithContext(ctx).Migrator().HasTable("knowledge_compile_docs") {
-		for _, column := range []string{"backlog_doc_ids", "inflight_doc_ids"} {
-			columnType, nullable, exists, err := columnDefinition(ctx, db, "knowledge_compile_docs", column)
-			if err != nil {
-				return fmt.Errorf("failed to verify knowledge_compile_docs.%s after migration: %w", column, err)
-			}
-			if exists && (normalizeColumnType(columnType) != "longtext" || nullable) {
-				return fmt.Errorf("knowledge_compile_docs.%s is %s nullable=%v after migration, want longtext NOT NULL", column, columnType, nullable)
-			}
-		}
-	}
-
 	return nil
 }
 

internal/dao/migration_integration_test.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"testing"
 
+	"ragflow/internal/entity"
+
 	"gorm.io/gorm"
 )
 
@@ -134,7 +136,7 @@ func TestModifyColumnTypesHealsLegacyTenantModelIDColumns_Integration(t *testing
 	}
 }
 
-func TestModifyColumnTypesWidensKnowledgeCompileBacklogs_Integration(t *testing.T) {
+func TestMigrateKnowledgeCompileQueues_Integration(t *testing.T) {
 	dsn := os.Getenv("RAGFLOW_TEST_MYSQL_DSN")
 	if dsn == "" {
 		t.Skip("RAGFLOW_TEST_MYSQL_DSN unset: this test needs a throwaway MySQL database")
@@ -155,22 +157,29 @@ func TestModifyColumnTypesWidensKnowledgeCompileBacklogs_Integration(t *testing.
 	})
 	ctx := context.Background()
 	assertModifyColumnTypesTablesAbsent(t, db.WithContext(ctx), "knowledge_compile_docs")
+	for _, model := range []interface{}{&entity.KnowledgeCompileBacklog{}, &entity.KnowledgeCompileInflight{}} {
+		if db.Migrator().HasTable(model) {
+			t.Fatalf("%T already exists; this test requires an empty database", model)
+		}
+	}
 	if err = db.WithContext(ctx).Exec(`CREATE TABLE knowledge_compile_docs (
 		dataset_id varchar(64) NOT NULL PRIMARY KEY,
-		backlog_doc_ids text NOT NULL,
-		inflight_doc_ids text NOT NULL
+		backlog_doc_ids longtext NOT NULL,
+		inflight_doc_ids longtext NOT NULL
 	)`).Error; err != nil {
 		t.Fatalf("failed to create legacy knowledge_compile_docs: %v", err)
 	}
 	t.Cleanup(func() {
-		if err := db.WithContext(context.Background()).Exec("DROP TABLE IF EXISTS knowledge_compile_docs").Error; err != nil {
-			t.Errorf("failed to drop knowledge_compile_docs test table: %v", err)
+		for _, table := range []string{"knowledge_compile_docs", "knowledge_compile_docs_backlog", "knowledge_compile_docs_inflight"} {
+			if err := db.WithContext(context.Background()).Migrator().DropTable(table); err != nil {
+				t.Errorf("failed to drop %s test table: %v", table, err)
+			}
 		}
 	})
 
-	legacyBacklog := knowledgeCompileBacklogFixture(480)
-	if len(legacyBacklog) >= 65535 {
-		t.Fatalf("legacy fixture is %d bytes, want less than TEXT's 65,535-byte limit", len(legacyBacklog))
+	legacyBacklog := knowledgeCompileBacklogFixture(600)
+	if len(legacyBacklog) <= 65535 {
+		t.Fatalf("fixture is %d bytes, want more than TEXT's 65,535-byte limit", len(legacyBacklog))
 	}
 	if err = db.WithContext(ctx).Exec(
 		"INSERT INTO knowledge_compile_docs (dataset_id, backlog_doc_ids, inflight_doc_ids) VALUES (?, ?, '[]')",
@@ -178,46 +187,34 @@ func TestModifyColumnTypesWidensKnowledgeCompileBacklogs_Integration(t *testing.
 	).Error; err != nil {
 		t.Fatalf("failed to seed legacy backlog: %v", err)
 	}
-	if err = modifyColumnTypes(ctx, db); err != nil {
-		t.Fatalf("modifyColumnTypes failed: %v", err)
+	if err := db.AutoMigrate(&entity.KnowledgeCompileBacklog{}, &entity.KnowledgeCompileInflight{}); err != nil {
+		t.Fatal(err)
+	}
+	if err = migrateKnowledgeCompileQueues(ctx, db); err != nil {
+		t.Fatalf("migrateKnowledgeCompileQueues failed: %v", err)
 	}
 
 	for _, column := range []string{"backlog_doc_ids", "inflight_doc_ids"} {
-		columnType, nullable, ok, err := columnDefinition(ctx, db, "knowledge_compile_docs", column)
-		if err != nil {
-			t.Fatalf("failed to read %s definition: %v", column, err)
-		}
-		if !ok || columnType != "longtext" || nullable {
-			t.Errorf("%s definition = %q nullable=%v exists=%v, want longtext NOT NULL", column, columnType, nullable, ok)
+		if db.Migrator().HasColumn(&entity.KnowledgeCompileDataset{}, column) {
+			t.Errorf("source column %s still exists", column)
 		}
 	}
-
-	var preserved string
-	if err = db.WithContext(ctx).Raw(
-		"SELECT backlog_doc_ids FROM knowledge_compile_docs WHERE dataset_id = ?", "kb1",
-	).Scan(&preserved).Error; err != nil {
-		t.Fatalf("failed to read migrated backlog: %v", err)
+	var entries []entity.KnowledgeCompileEntry
+	if err := db.Table((entity.KnowledgeCompileBacklog{}).TableName()).Order("queue_position").Find(&entries).Error; err != nil {
+		t.Fatal(err)
 	}
-	if preserved != legacyBacklog {
-		t.Fatal("migration changed the existing backlog")
+	if len(entries) != 600 {
+		t.Fatalf("migrated %d events, want 600", len(entries))
 	}
-
-	oversizedBacklog := knowledgeCompileBacklogFixture(600)
-	if len(oversizedBacklog) <= 65535 {
-		t.Fatalf("oversized fixture is %d bytes, want more than TEXT's 65,535-byte limit", len(oversizedBacklog))
+	for i, entry := range entries {
+		if entry.DocID != fmt.Sprintf("%032d", i) || entry.QueuePosition != int64(i+1) || entry.EventType != "doc_completed" || len(entry.TaskTypes) != 1 || entry.TaskTypes[0] != "Graph" {
+			t.Fatalf("unexpected event %d: %+v", i, entry)
+		}
 	}
-	if err = db.WithContext(ctx).Exec(
-		"UPDATE knowledge_compile_docs SET backlog_doc_ids = ? WHERE dataset_id = ?",
-		oversizedBacklog, "kb1",
-	).Error; err != nil {
-		t.Fatalf("failed to store backlog larger than MySQL TEXT: %v", err)
-	}
-	if err = db.WithContext(ctx).Raw(
-		"SELECT backlog_doc_ids FROM knowledge_compile_docs WHERE dataset_id = ?", "kb1",
-	).Scan(&preserved).Error; err != nil {
-		t.Fatalf("failed to read oversized backlog: %v", err)
-	}
-	if preserved != oversizedBacklog {
-		t.Fatal("oversized backlog was truncated")
+	for _, table := range []string{"knowledge_compile_docs_backlog", "knowledge_compile_docs_inflight"} {
+		var count int64
+		if err := db.Raw("SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = DATABASE() AND table_name = ? AND non_unique = 0", table).Scan(&count).Error; err != nil || count != 0 {
+			t.Fatalf("%s must have no primary key or unique indexes: count=%d err=%v", table, count, err)
+		}
 	}
 }

internal/dao/migration_test.go

@@ -30,19 +30,24 @@ import (
 
 const tenantModelIDWidth = 32
 
-func TestKnowledgeCompileBacklogColumnsUseLongText(t *testing.T) {
-	parsed, err := schema.Parse(&entity.KnowledgeCompileDataset{}, &sync.Map{}, schema.NamingStrategy{})
-	if err != nil {
-		t.Fatalf("parse KnowledgeCompileDataset schema: %v", err)
-	}
-
-	for _, name := range []string{"BacklogDocIDs", "InflightDocIDs"} {
-		field := parsed.FieldsByName[name]
-		if field == nil {
-			t.Fatalf("missing field %s", name)
+func TestKnowledgeCompileQueuesHaveNoPrimaryKey(t *testing.T) {
+	for _, model := range []interface{}{&entity.KnowledgeCompileBacklog{}, &entity.KnowledgeCompileInflight{}} {
+		parsed, err := schema.Parse(model, &sync.Map{}, schema.NamingStrategy{})
+		if err != nil {
+			t.Fatalf("parse %T schema: %v", model, err)
 		}
-		if field.DataType != "longtext" {
-			t.Errorf("%s database type = %q, want longtext", name, field.DataType)
+		if len(parsed.PrimaryFields) != 0 || parsed.FieldsByName["ID"] != nil {
+			t.Fatalf("%s must have no primary key or ID", parsed.Table)
+		}
+		indexes := parsed.ParseIndexes()
+		if len(indexes) != 1 {
+			t.Fatalf("%s indexes = %v, want one queue index", parsed.Table, indexes)
+		}
+		for _, index := range indexes {
+			if index.Class == "UNIQUE" || len(index.Fields) != 2 ||
+				index.Fields[0].DBName != "dataset_id" || index.Fields[1].DBName != "queue_position" {
+				t.Fatalf("unexpected queue index: %+v", index)
+			}
 		}
 	}
 }

internal/entity/knowledge_compile_doc.go

@@ -29,22 +29,13 @@ const (
 
 // KnowledgeCompileDataset is the MySQL scheduling row for the dataset-level
 // post-processing consumer (knowledge_compile_design.md §11.4, Option E). It is
-// the scheduling system of record: backlog_doc_ids holds the not-yet-processed
-// doc entries for the KB, inflight_doc_ids the ones a worker has claimed (the
-// closed batch), and the claim_* fields the owner/lease used for crash
-// recovery. NATS notify is only a wake-up; same-KB serialization comes from
-// these rows, not from the broker.
-//
-// The *_doc_ids columns store JSON arrays of knowledge_compile.BacklogEntry
-// (doc_id + event_type + seq) as LONGTEXT so the consumer can re-apply the same
-// out-of-order / tombstone guards as the broker-based design without re-reading
-// the queue, even when a backlog exceeds MySQL TEXT's 64 KiB limit.
+// the scheduling system of record together with KnowledgeCompileBacklog and
+// KnowledgeCompileInflight. The claim_* fields hold the owner/lease used for
+// crash recovery. NATS notify is only a wake-up; same-KB serialization comes
+// from locking this row before changing either queue, not from the broker.
 type KnowledgeCompileDataset struct {
-	DatasetID string `gorm:"primaryKey;column:dataset_id;size:64" json:"dataset_id"`
-	TenantID  string `gorm:"column:tenant_id;size:64;not null;default:''" json:"tenant_id"`
-	// No DDL default is set because the application writes "[]" explicitly.
-	BacklogDocIDs  string     `gorm:"column:backlog_doc_ids;type:longtext;not null" json:"backlog_doc_ids"`
-	InflightDocIDs string     `gorm:"column:inflight_doc_ids;type:longtext;not null" json:"inflight_doc_ids"`
+	DatasetID      string     `gorm:"primaryKey;column:dataset_id;size:64" json:"dataset_id"`
+	TenantID       string     `gorm:"column:tenant_id;size:64;not null;default:''" json:"tenant_id"`
 	ClaimOwner     string     `gorm:"column:claim_owner;size:64;not null;default:''" json:"claim_owner"`
 	ClaimToken     string     `gorm:"column:claim_token;size:64;not null;default:''" json:"claim_token"`
 	ClaimExpiresAt *time.Time `gorm:"column:claim_expires_at;default:null" json:"claim_expires_at"`
@@ -70,3 +61,27 @@ type KnowledgeCompileDataset struct {
 
 // TableName pins the scheduling table name.
 func (KnowledgeCompileDataset) TableName() string { return "knowledge_compile_docs" }
+
+// KnowledgeCompileEntry is an ordered queue event. Duplicate document events
+// are allowed; neither queue has a primary key. Positions are allocated while
+// holding the dataset scheduling row lock.
+type KnowledgeCompileEntry struct {
+	DatasetID     string   `gorm:"column:dataset_id;size:64;not null;index:,composite:queue,priority:1" json:"dataset_id"`
+	DocID         string   `gorm:"column:doc_id;size:64;not null" json:"doc_id"`
+	QueuePosition int64    `gorm:"column:queue_position;not null;index:,composite:queue,priority:2" json:"queue_position"`
+	EventType     string   `gorm:"column:event_type;size:32;not null" json:"event_type"`
+	Variants      []string `gorm:"column:variants;type:longtext;serializer:json" json:"variants,omitempty"`
+	TaskTypes     []string `gorm:"column:task_types;type:longtext;serializer:json" json:"task_types,omitempty"`
+}
+
+type KnowledgeCompileBacklog struct {
+	KnowledgeCompileEntry `gorm:"embedded"`
+}
+
+func (KnowledgeCompileBacklog) TableName() string { return "knowledge_compile_docs_backlog" }
+
+type KnowledgeCompileInflight struct {
+	KnowledgeCompileEntry `gorm:"embedded"`
+}
+
+func (KnowledgeCompileInflight) TableName() string { return "knowledge_compile_docs_inflight" }

internal/handler/compilation_status_test.go

@@ -45,6 +45,8 @@ func setupCompilationStatusHandlerDB(t *testing.T) *gorm.DB {
 	if err := db.AutoMigrate(
 		&entity.Knowledgebase{},
 		&entity.KnowledgeCompileDataset{},
+		&entity.KnowledgeCompileBacklog{},
+		&entity.KnowledgeCompileInflight{},
 		&entity.UserTenant{},
 	); err != nil {
 		t.Fatalf("failed to migrate test schema: %v", err)
@@ -158,16 +160,24 @@ func TestCompilationStatusHandler_FullOutput(t *testing.T) {
 	insertCompilationStatusHandlerKB(t, "kb-status-full", "user-1")
 
 	row := entity.KnowledgeCompileDataset{
-		DatasetID:      "kb-status-full",
-		TenantID:       "user-1",
-		BacklogDocIDs:  `[{"doc_id":"d2","event_type":"completed","task_types":["Graph"]}]`,
-		InflightDocIDs: `[{"doc_id":"d1","event_type":"completed","task_types":["Graph"]}]`,
-		State:          entity.DatasetStatePending,
-		ErrorMsg:       "merge failed: boom",
+		DatasetID: "kb-status-full",
+		TenantID:  "user-1",
+		State:     entity.DatasetStatePending,
+		ErrorMsg:  "merge failed: boom",
 	}
 	if err := db.Create(&row).Error; err != nil {
 		t.Fatalf("insert scheduling row: %v", err)
 	}
+	if err := db.Create(&entity.KnowledgeCompileBacklog{KnowledgeCompileEntry: entity.KnowledgeCompileEntry{
+		DatasetID: row.DatasetID, DocID: "d2", QueuePosition: 1, EventType: "completed", TaskTypes: []string{"Graph"},
+	}}).Error; err != nil {
+		t.Fatalf("insert backlog: %v", err)
+	}
+	if err := db.Create(&entity.KnowledgeCompileInflight{KnowledgeCompileEntry: entity.KnowledgeCompileEntry{
+		DatasetID: row.DatasetID, DocID: "d1", QueuePosition: 1, EventType: "completed", TaskTypes: []string{"Graph"},
+	}}).Error; err != nil {
+		t.Fatalf("insert inflight: %v", err)
+	}
 
 	status, body := getCompilationStatusWithKind(t, newCompilationStatusHandlerRouter(), "kb-status-full", "graph")
 	if status != http.StatusOK {
@@ -203,11 +213,9 @@ func TestCompilationStatusHandler_Unauthorized(t *testing.T) {
 	// KB owner to "other-owner".
 	insertCompilationStatusHandlerKB(t, "kb-status-forbidden", "other-owner")
 	if err := db.Create(&entity.KnowledgeCompileDataset{
-		DatasetID:      "kb-status-forbidden",
-		TenantID:       "other-owner",
-		BacklogDocIDs:  "[]",
-		InflightDocIDs: "[]",
-		State:          entity.DatasetStateRunning,
+		DatasetID: "kb-status-forbidden",
+		TenantID:  "other-owner",
+		State:     entity.DatasetStateRunning,
 	}).Error; err != nil {
 		t.Fatalf("insert scheduling row: %v", err)
 	}

internal/ingestion/knowledge_compile/consumer.go

@@ -593,9 +593,9 @@ func (c *Consumer) processBatch(ctx context.Context, tenant, kb, token string, e
 	// land in the same batch, and completions/deletions can also arrive out of
 	// order across batches. The LAST event for a doc (by backlog append order)
 	// decides its fate, so we must not delete a doc that was re-ingested after
-	// its deletion, nor merge a completion shadowed by a later deletion. No
-	// sequence number is needed: the backlog is an ordered log and the last
-	// write wins, which is exactly the re-parse (completed last) vs delete
+	// its deletion, nor merge a completion shadowed by a later deletion. Queue
+	// positions preserve the backlog's append order and the last write wins,
+	// which is exactly the re-parse (completed last) vs delete
 	// (deleted last) semantics we want.
 	type docState struct {
 		winEvent EventType

internal/ingestion/knowledge_compile/event.go

@@ -18,12 +18,13 @@
 //
 // Pipeline (KnowledgeCompiler, per document) writes compiled chunks with
 // available_int=0. Its completion/deletion is recorded by appending a
-// BacklogEntry to the KB's durable MySQL scheduling row
-// (knowledge_compile_docs), then waking idle workers over NATS. A cluster of
+// BacklogEntry to knowledge_compile_docs_backlog under the KB's scheduling row
+// lock (knowledge_compile_docs), then waking idle workers over NATS. A cluster of
 // competing workers claims a closed batch (backlog -> inflight) per KB, runs
 // dataset-level dedup on that batch, and writes the merged dataset-level
-// products with available_int=1. The MySQL row — not the broker — is the
-// scheduling system of record and the source of same-KB serialization.
+// products with available_int=1. The MySQL scheduling row and its backlog and
+// inflight tables are the scheduling system of record; the row lock serializes
+// same-KB processing.
 package knowledge_compile
 
 import (

internal/ingestion/knowledge_compile/scheduler.go

@@ -17,6 +17,7 @@ package knowledge_compile
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -56,7 +57,7 @@ const (
 // §11.4). It carries the doc id plus the original event kind so the consumer
 // can re-apply the same tombstone handling as the broker-based design without
 // re-reading the queue. The backlog is an ordered log; the last event for a
-// given doc (by append order) wins, so no sequence number is needed.
+// given doc (by queue position) wins.
 //
 // Variants records the compile types (tree/structure/wiki/mindmap) produced by
 // the doc-level compile for this doc, so the consumer can group a claimed batch
@@ -167,27 +168,6 @@ func newScheduler(db *gorm.DB, mq engine.MessageQueue, holder string, ttl time.D
 	return &mysqlScheduler{db: db, mq: mq, holder: holder, leaseTTL: ttl, claimBatchSize: defaultClaimBatch}
 }
 
-// ---- JSON helpers (the *_doc_ids columns are TEXT holding []BacklogEntry) ----
-
-func marshalEntries(es []BacklogEntry) string {
-	if len(es) == 0 {
-		return "[]"
-	}
-	b, _ := json.Marshal(es)
-	return string(b)
-}
-
-func parseEntries(s string) []BacklogEntry {
-	if s == "" || s == "[]" {
-		return nil
-	}
-	var es []BacklogEntry
-	if err := json.Unmarshal([]byte(s), &es); err != nil || len(es) == 0 {
-		return nil
-	}
-	return es
-}
-
 func minInt(a, b int) int {
 	if a < b {
 		return a
@@ -213,7 +193,54 @@ func (s *mysqlScheduler) Provision(ctx context.Context) error {
 	if s.db == nil {
 		return nil
 	}
-	return s.db.WithContext(ctx).AutoMigrate(&entity.KnowledgeCompileDataset{})
+	return s.db.WithContext(ctx).AutoMigrate(
+		&entity.KnowledgeCompileDataset{},
+		&entity.KnowledgeCompileBacklog{},
+		&entity.KnowledgeCompileInflight{},
+	)
+}
+
+// appendQueueEntries requires the caller to hold the dataset scheduling row
+// lock. Positions preserve append order without deduplicating document events.
+func appendQueueEntries(tx *gorm.DB, table, datasetID string, entries []entity.KnowledgeCompileEntry) error {
+	if len(entries) == 0 {
+		return nil
+	}
+	var lastPosition int64
+	if err := tx.Table(table).Where("dataset_id = ?", datasetID).
+		Select("COALESCE(MAX(queue_position), 0)").Scan(&lastPosition).Error; err != nil {
+		return err
+	}
+	for i := range entries {
+		entries[i].DatasetID = datasetID
+		entries[i].QueuePosition = lastPosition + int64(i) + 1
+	}
+	return tx.Table(table).CreateInBatches(&entries, defaultClaimBatch).Error
+}
+
+// requeueInflight appends the inflight queue after the existing backlog, as
+// cancellation and lease recovery do. The caller holds the dataset row lock.
+func requeueInflight(tx *gorm.DB, datasetID string) (int64, error) {
+	backlogTable := (entity.KnowledgeCompileBacklog{}).TableName()
+	inflightTable := (entity.KnowledgeCompileInflight{}).TableName()
+	var lastPosition int64
+	if err := tx.Table(backlogTable).Where("dataset_id = ?", datasetID).
+		Select("COALESCE(MAX(queue_position), 0)").Scan(&lastPosition).Error; err != nil {
+		return 0, err
+	}
+	// Both table names are fixed entity names. Copying within SQL also bounds
+	// memory when recovering an unusually large inflight queue.
+	res := tx.Exec("INSERT INTO "+backlogTable+
+		" (dataset_id, doc_id, queue_position, event_type, variants, task_types) "+
+		"SELECT dataset_id, doc_id, queue_position + ?, event_type, variants, task_types FROM "+inflightTable+
+		" WHERE dataset_id = ?", lastPosition, datasetID)
+	if res.Error != nil {
+		return 0, res.Error
+	}
+	if err := tx.Where("dataset_id = ?", datasetID).Delete(&entity.KnowledgeCompileInflight{}).Error; err != nil {
+		return 0, err
+	}
+	return res.RowsAffected, nil
 }
 
 // Publish appends one doc event to the dataset's durable backlog and wakes idle
@@ -257,10 +284,8 @@ func (s *mysqlScheduler) publishEntry(ctx context.Context, tenantID, datasetID s
 		// dataset_id find it (M19: a raw-string Where alone leaves DatasetID
 		// blank on the created row).
 		row := entity.KnowledgeCompileDataset{
-			DatasetID:      datasetID,
-			TenantID:       tenantID,
-			BacklogDocIDs:  "[]",
-			InflightDocIDs: "[]",
+			DatasetID: datasetID,
+			TenantID:  tenantID,
 		}
 		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
 			Where("dataset_id = ?", datasetID).
@@ -270,9 +295,10 @@ func (s *mysqlScheduler) publishEntry(ctx context.Context, tenantID, datasetID s
 		if row.TenantID == "" {
 			row.TenantID = tenantID
 		}
-		backlog := parseEntries(row.BacklogDocIDs)
-		backlog = append(backlog, entry)
-		row.BacklogDocIDs = marshalEntries(backlog)
+		if err := appendQueueEntries(tx, (entity.KnowledgeCompileBacklog{}).TableName(), datasetID,
+			[]entity.KnowledgeCompileEntry{{DocID: entry.DocID, EventType: entry.EventType, Variants: entry.Variants, TaskTypes: entry.TaskTypes}}); err != nil {
+			return err
+		}
 		// Surface a pending state unless a worker is already running (a live lease
 		// means the consumer is mid-merge on this KB; a pending transition would
 		// wrongly hide that). completed -> pending when new work arrives.
@@ -312,9 +338,7 @@ func (s *mysqlScheduler) WithDatasetLock(ctx context.Context, datasetID string,
 	}
 	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		row := entity.KnowledgeCompileDataset{
-			DatasetID:      datasetID,
-			BacklogDocIDs:  "[]",
-			InflightDocIDs: "[]",
+			DatasetID: datasetID,
 		}
 		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
 			Where("dataset_id = ?", datasetID).
@@ -344,16 +368,32 @@ func (s *mysqlScheduler) claimRow(ctx context.Context, tx *gorm.DB, datasetID st
 	if liveLease {
 		return ClaimResult{}, false, nil // a live lease means some worker is already processing this KB
 	}
-	backlog := parseEntries(row.BacklogDocIDs)
+	var backlog []entity.KnowledgeCompileEntry
+	if err := tx.Table((entity.KnowledgeCompileBacklog{}).TableName()).
+		Where("dataset_id = ?", datasetID).Order("queue_position ASC").
+		Limit(s.claimBatchSize).Find(&backlog).Error; err != nil {
+		return ClaimResult{}, false, err
+	}
 	if len(backlog) == 0 {
 		return ClaimResult{}, false, nil
 	}
-	n := minInt(s.claimBatchSize, len(backlog))
-	batch := backlog[:n]
-	inflight := parseEntries(row.InflightDocIDs)
-	inflight = append(inflight, batch...)
-	row.BacklogDocIDs = marshalEntries(backlog[n:])
-	row.InflightDocIDs = marshalEntries(inflight)
+	positions := make([]int64, len(backlog))
+	batch := make([]BacklogEntry, len(backlog))
+	for i, entry := range backlog {
+		positions[i] = entry.QueuePosition
+		batch[i] = BacklogEntry{DocID: entry.DocID, EventType: entry.EventType, Variants: entry.Variants, TaskTypes: entry.TaskTypes}
+	}
+	if err := appendQueueEntries(tx, (entity.KnowledgeCompileInflight{}).TableName(), datasetID, backlog); err != nil {
+		return ClaimResult{}, false, err
+	}
+	if err := tx.Where("dataset_id = ? AND queue_position IN ?", datasetID, positions).
+		Delete(&entity.KnowledgeCompileBacklog{}).Error; err != nil {
+		return ClaimResult{}, false, err
+	}
+	var remaining int64
+	if err := tx.Model(&entity.KnowledgeCompileBacklog{}).Where("dataset_id = ?", datasetID).Count(&remaining).Error; err != nil {
+		return ClaimResult{}, false, err
+	}
 	row.ClaimOwner = s.holder
 	row.ClaimToken = generateHolder()
 	exp := now.Add(s.leaseTTL)
@@ -370,7 +410,7 @@ func (s *mysqlScheduler) claimRow(ctx context.Context, tx *gorm.DB, datasetID st
 		zap.String("dataset_id", row.DatasetID),
 		zap.String("tenant_id", row.TenantID),
 		zap.Int("batch_size", len(batch)),
-		zap.Int("backlog_remaining", len(backlog)-n),
+		zap.Int64("backlog_remaining", remaining),
 		zap.String("token", row.ClaimToken),
 		zap.Any("batch", batch))
 	return ClaimResult{DatasetID: row.DatasetID, TenantID: row.TenantID, Entries: batch, Token: row.ClaimToken}, true, nil
@@ -410,15 +450,30 @@ func (s *mysqlScheduler) Ack(ctx context.Context, datasetID, token string, batch
 		if row.ClaimToken != token {
 			return ErrClaimTokenMismatch
 		}
-		inflight := parseEntries(row.InflightDocIDs)
-		inflight = removeEntries(inflight, batch)
-		row.InflightDocIDs = marshalEntries(inflight)
-		if len(inflight) == 0 {
+		if len(batch) > 0 {
+			pairs := make([][]interface{}, len(batch))
+			for i, entry := range batch {
+				pairs[i] = []interface{}{entry.DocID, entry.EventType}
+			}
+			if err := tx.Where("dataset_id = ? AND (doc_id, event_type) IN ?", datasetID, pairs).
+				Delete(&entity.KnowledgeCompileInflight{}).Error; err != nil {
+				return err
+			}
+		}
+		var inflight int64
+		if err := tx.Model(&entity.KnowledgeCompileInflight{}).Where("dataset_id = ?", datasetID).Count(&inflight).Error; err != nil {
+			return err
+		}
+		if inflight == 0 {
 			row.ClaimOwner = ""
 			row.ClaimToken = ""
 			row.ClaimExpiresAt = nil
 		}
-		remaining = len(parseEntries(row.BacklogDocIDs))
+		var backlog int64
+		if err := tx.Model(&entity.KnowledgeCompileBacklog{}).Where("dataset_id = ?", datasetID).Count(&backlog).Error; err != nil {
+			return err
+		}
+		remaining = int(backlog)
 		// Backlog still has work -> pending; drained to empty -> completed.
 		if remaining > 0 {
 			row.State = DatasetStatePending
@@ -553,14 +608,12 @@ func (s *mysqlScheduler) CancelInflight(ctx context.Context, datasetID, token st
 		if !liveLease {
 			return nil
 		}
-		inflight := parseEntries(row.InflightDocIDs)
-		backlog := parseEntries(row.BacklogDocIDs)
-		backlog = append(backlog, inflight...)
-		row.InflightDocIDs = "[]"
+		if _, err := requeueInflight(tx, datasetID); err != nil {
+			return err
+		}
 		row.ClaimOwner = ""
 		row.ClaimToken = ""
 		row.ClaimExpiresAt = nil
-		row.BacklogDocIDs = marshalEntries(backlog)
 		row.State = DatasetStatePending
 		return tx.Save(&row).Error
 	})
@@ -601,10 +654,14 @@ func (s *mysqlScheduler) TryClaim(ctx context.Context) (ClaimResult, bool, error
 	}
 	var res ClaimResult
 	var acquired bool
+	// Candidate reads precede the dataset lock. Read committed ensures the
+	// queue reads after that lock see events committed while it was acquired.
 	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		now := time.Now()
 		// 1) a ready dataset (backlog, no live lease) — claim it directly.
-		if id, ok := s.findClaimableID(ctx, tx, now); ok {
+		if id, ok, err := s.findClaimableID(ctx, tx, now); err != nil {
+			return err
+		} else if ok {
 			cr, ok, e := s.claimRow(ctx, tx, id)
 			if e != nil {
 				return e
@@ -624,7 +681,7 @@ func (s *mysqlScheduler) TryClaim(ctx context.Context) (ClaimResult, bool, error
 			return nil
 		}
 		return nil
-	})
+	}, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
 	if err != nil {
 		return ClaimResult{}, false, err
 	}
@@ -632,19 +689,18 @@ func (s *mysqlScheduler) TryClaim(ctx context.Context) (ClaimResult, bool, error
 }
 
 // findClaimableID returns one dataset id with non-empty backlog and no live
-// lease, or ok=false when none exists. It runs inside caller's transaction so
-// the returned id is still locked when the caller claims it.
-func (s *mysqlScheduler) findClaimableID(ctx context.Context, tx *gorm.DB, now time.Time) (string, bool) {
+// lease, or ok=false when none exists. claimRow locks and rechecks the selected
+// dataset before moving any queue entries.
+func (s *mysqlScheduler) findClaimableID(ctx context.Context, tx *gorm.DB, now time.Time) (string, bool, error) {
 	var id string
+	backlog := tx.Model(&entity.KnowledgeCompileBacklog{}).Select("1").
+		Where("dataset_id = knowledge_compile_docs.dataset_id")
 	err := tx.Model(&entity.KnowledgeCompileDataset{}).
-		Where("(claim_expires_at IS NULL OR claim_expires_at <= ?) AND backlog_doc_ids <> '[]' AND backlog_doc_ids <> '' AND backlog_doc_ids IS NOT NULL", now).
+		Where("(claim_expires_at IS NULL OR claim_expires_at <= ?) AND EXISTS (?)", now, backlog).
 		Order("priority DESC, updated_at ASC").
 		Limit(1).
 		Pluck("dataset_id", &id).Error
-	if err != nil || id == "" {
-		return "", false
-	}
-	return id, true
+	return id, id != "", err
 }
 
 // reclaimOne moves one expired inflight batch back to backlog (crash recovery)
@@ -653,8 +709,10 @@ func (s *mysqlScheduler) findClaimableID(ctx context.Context, tx *gorm.DB, now t
 // caller claims it.
 func (s *mysqlScheduler) reclaimOne(ctx context.Context, tx *gorm.DB, now time.Time) (string, bool, error) {
 	var rows []entity.KnowledgeCompileDataset
+	inflight := tx.Model(&entity.KnowledgeCompileInflight{}).Select("1").
+		Where("dataset_id = knowledge_compile_docs.dataset_id")
 	if err := tx.Model(&entity.KnowledgeCompileDataset{}).
-		Where("claim_expires_at IS NOT NULL AND claim_expires_at <= ? AND inflight_doc_ids <> '[]' AND inflight_doc_ids <> ''", now).
+		Where("claim_expires_at IS NOT NULL AND claim_expires_at <= ? AND EXISTS (?)", now, inflight).
 		Find(&rows).Error; err != nil {
 		return "", false, err
 	}
@@ -670,14 +728,13 @@ func (s *mysqlScheduler) reclaimOne(ctx context.Context, tx *gorm.DB, now time.T
 		if cur.ClaimExpiresAt != nil && cur.ClaimExpiresAt.After(now) {
 			continue
 		}
-		inflight := parseEntries(cur.InflightDocIDs)
-		if len(inflight) == 0 {
+		reclaimed, err := requeueInflight(tx, cur.DatasetID)
+		if err != nil {
+			return "", false, err
+		}
+		if reclaimed == 0 {
 			continue
 		}
-		backlog := parseEntries(cur.BacklogDocIDs)
-		backlog = append(backlog, inflight...)
-		cur.BacklogDocIDs = marshalEntries(backlog)
-		cur.InflightDocIDs = "[]"
 		cur.ClaimOwner = ""
 		cur.ClaimToken = ""
 		cur.ClaimExpiresAt = nil
@@ -688,7 +745,7 @@ func (s *mysqlScheduler) reclaimOne(ctx context.Context, tx *gorm.DB, now time.T
 		common.Warn("knowledge_compile: reclaimed expired inflight lease",
 			zap.String("dataset_id", cur.DatasetID),
 			zap.String("tenant_id", cur.TenantID),
-			zap.Int("reclaimed_entries", len(inflight)))
+			zap.Int64("reclaimed_entries", reclaimed))
 		return cur.DatasetID, true, nil
 	}
 	return "", false, nil

internal/ingestion/knowledge_compile/scheduler_integration_test.go

@@ -0,0 +1,159 @@
+//go:build integration
+
+package knowledge_compile
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"ragflow/internal/entity"
+
+	"gorm.io/driver/mysql"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+)
+
+func TestMySQLSchedulerQueues_Integration(t *testing.T) {
+	dsn := os.Getenv("RAGFLOW_TEST_MYSQL_DSN")
+	if dsn == "" {
+		t.Skip("RAGFLOW_TEST_MYSQL_DSN unset: requires an empty MySQL test database")
+	}
+	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pool, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { pool.Close() })
+	models := []interface{}{&entity.KnowledgeCompileDataset{}, &entity.KnowledgeCompileBacklog{}, &entity.KnowledgeCompileInflight{}}
+	for _, model := range models {
+		if db.Migrator().HasTable(model) {
+			t.Fatalf("%T already exists; this test requires an empty database", model)
+		}
+	}
+	t.Cleanup(func() {
+		for _, model := range models {
+			if err := db.Migrator().DropTable(model); err != nil {
+				t.Errorf("drop %T: %v", model, err)
+			}
+		}
+	})
+	s := newScheduler(db, nil, "worker1", time.Minute)
+	if err := s.Provision(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Publish(t.Context(), "tenant", "kb", "first", "doc_completed", []string{"wiki"}, []string{"Wiki"}); err != nil {
+		t.Fatal(err)
+	}
+	var sqlLog bytes.Buffer
+	dryRun := db.Session(&gorm.Session{
+		DryRun: true,
+		Logger: logger.New(log.New(&sqlLog, "", 0), logger.Config{LogLevel: logger.Info}),
+	})
+	if _, _, err := s.findClaimableID(t.Context(), dryRun, time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("MySQL candidate query: %s", sqlLog.String())
+
+	const publishers = 4
+	const eventsPerPublisher = 16
+	errorsCh := make(chan error, publishers)
+	var writers sync.WaitGroup
+	for worker := 0; worker < publishers; worker++ {
+		writers.Add(1)
+		go func() {
+			defer writers.Done()
+			for event := 0; event < eventsPerPublisher; event++ {
+				if err := s.Publish(t.Context(), "tenant", "kb", fmt.Sprintf("doc-%d-%d", worker, event), "doc_completed", []string{"wiki"}, []string{"Wiki"}); err != nil {
+					errorsCh <- err
+					return
+				}
+			}
+		}()
+	}
+	writers.Wait()
+	close(errorsCh)
+	for err := range errorsCh {
+		t.Fatal(err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 65, 0)
+	var entries []entity.KnowledgeCompileEntry
+	if err := db.Table((entity.KnowledgeCompileBacklog{}).TableName()).Order("queue_position").Find(&entries).Error; err != nil {
+		t.Fatal(err)
+	}
+	for i, entry := range entries {
+		if entry.QueuePosition != int64(i+1) {
+			t.Fatalf("concurrent publishers reused queue positions: %+v", entries)
+		}
+	}
+
+	type claimOutcome struct {
+		claim ClaimResult
+		ok    bool
+		err   error
+	}
+	start := make(chan struct{})
+	claimed := make(chan claimOutcome, 2)
+	for _, scheduler := range []*mysqlScheduler{s, newScheduler(db, nil, "worker2", time.Minute)} {
+		go func() {
+			<-start
+			claim, ok, err := scheduler.TryClaim(t.Context())
+			claimed <- claimOutcome{claim, ok, err}
+		}()
+	}
+	close(start)
+	var claim ClaimResult
+	acquired := 0
+	for range 2 {
+		outcome := <-claimed
+		if outcome.err != nil {
+			t.Fatal(outcome.err)
+		}
+		if outcome.ok {
+			acquired++
+			claim = outcome.claim
+		}
+	}
+	if acquired != 1 || len(claim.Entries) != defaultClaimBatch {
+		t.Fatalf("concurrent claims: acquired=%d claim=%+v", acquired, claim)
+	}
+	assertSQLQueueCounts(t, s, "kb", 33, 32)
+	if remaining, err := s.Ack(t.Context(), "kb", claim.Token, claim.Entries); err != nil || remaining != 33 {
+		t.Fatalf("ack remaining=%d want=33 err=%v", remaining, err)
+	}
+	second, ok, err := s.TryClaim(t.Context())
+	if err != nil || !ok || len(second.Entries) != defaultClaimBatch {
+		t.Fatalf("second claim=%+v ok=%v err=%v", second, ok, err)
+	}
+	if remaining, err := s.Ack(t.Context(), "kb", second.Token, second.Entries); err != nil || remaining != 1 {
+		t.Fatalf("second ack remaining=%d want=1 err=%v", remaining, err)
+	}
+	lastClaim, ok, err := s.TryClaim(t.Context())
+	if err != nil || !ok || len(lastClaim.Entries) != 1 {
+		t.Fatalf("last claim=%+v ok=%v err=%v", lastClaim, ok, err)
+	}
+	if err := db.Model(&entity.KnowledgeCompileDataset{}).Where("dataset_id = ?", "kb").
+		Update("claim_expires_at", time.Now().Add(-time.Minute)).Error; err != nil {
+		t.Fatal(err)
+	}
+	recovered, ok, err := s.TryClaim(t.Context())
+	if err != nil || !ok || recovered.Token == lastClaim.Token || !reflect.DeepEqual(recovered.Entries, lastClaim.Entries) {
+		t.Fatalf("recovered=%+v previous=%+v ok=%v err=%v", recovered, lastClaim, ok, err)
+	}
+	if _, err := s.Ack(t.Context(), "kb", lastClaim.Token, lastClaim.Entries); !errors.Is(err, ErrClaimTokenMismatch) {
+		t.Fatalf("stale acknowledgment error=%v", err)
+	}
+	if _, err := s.Ack(t.Context(), "kb", recovered.Token, recovered.Entries); err != nil {
+		t.Fatal(err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 0, 0)
+}

internal/ingestion/knowledge_compile/scheduler_sql_test.go

@@ -0,0 +1,160 @@
+package knowledge_compile
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+	"time"
+
+	"ragflow/internal/entity"
+
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+)
+
+func newSQLTestScheduler(t *testing.T) *mysqlScheduler {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pool, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { pool.Close() })
+	scheduler := newScheduler(db, nil, "worker", time.Minute)
+	if err := scheduler.Provision(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	return scheduler
+}
+
+func assertSQLQueueCounts(t *testing.T, s *mysqlScheduler, datasetID string, backlog, inflight int64) {
+	t.Helper()
+	for _, queue := range []struct {
+		model interface{}
+		want  int64
+	}{
+		{&entity.KnowledgeCompileBacklog{}, backlog},
+		{&entity.KnowledgeCompileInflight{}, inflight},
+	} {
+		var count int64
+		if err := s.db.Model(queue.model).Where("dataset_id = ?", datasetID).Count(&count).Error; err != nil || count != queue.want {
+			t.Fatalf("%T count=%d want=%d err=%v", queue.model, count, queue.want, err)
+		}
+	}
+}
+
+func TestSQLSchedulerPreservesDuplicatesAndClosedBatches(t *testing.T) {
+	s := newSQLTestScheduler(t)
+	s.claimBatchSize = 2
+	entries := []BacklogEntry{
+		{DocID: "same", EventType: "doc_completed", Variants: []string{"wiki"}, TaskTypes: []string{"Wiki"}},
+		{DocID: "same", EventType: "doc_completed", Variants: []string{"wiki"}, TaskTypes: []string{"Wiki"}},
+		{DocID: "same", EventType: "doc_deleted", Variants: []string{"wiki"}, TaskTypes: []string{"Wiki"}},
+	}
+	for _, entry := range entries {
+		if err := s.Publish(t.Context(), "tenant", "kb", entry.DocID, entry.EventType, entry.Variants, entry.TaskTypes); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := s.Publish(t.Context(), "other-tenant", "other-kb", "same", "doc_completed", nil, nil); err != nil {
+		t.Fatal(err)
+	}
+	first, ok, err := s.Claim(t.Context(), "kb")
+	if err != nil || !ok || !reflect.DeepEqual(first.Entries, entries[:2]) {
+		t.Fatalf("claim=%+v ok=%v err=%v", first, ok, err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 1, 2)
+	if _, acquired, err := s.Claim(t.Context(), "kb"); err != nil || acquired {
+		t.Fatalf("live dataset was claimed twice: acquired=%v err=%v", acquired, err)
+	}
+	if _, err := s.Ack(t.Context(), "kb", "stale-token", first.Entries); !errors.Is(err, ErrClaimTokenMismatch) {
+		t.Fatalf("stale ack error=%v", err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 1, 2)
+	// A new completion for the same document must survive acknowledgment of
+	// the earlier completions, even when all event fields are identical.
+	if err := s.Publish(t.Context(), "tenant", "kb", "same", "doc_completed", []string{"wiki"}, []string{"Wiki"}); err != nil {
+		t.Fatal(err)
+	}
+	if remaining, err := s.Ack(t.Context(), "kb", first.Token, first.Entries); err != nil || remaining != 2 {
+		t.Fatalf("ack remaining=%d err=%v", remaining, err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 2, 0)
+	second, ok, err := s.Claim(t.Context(), "kb")
+	want := []BacklogEntry{entries[2], entries[0]}
+	if err != nil || !ok || !reflect.DeepEqual(second.Entries, want) {
+		t.Fatalf("second claim=%+v want=%+v ok=%v err=%v", second, want, ok, err)
+	}
+	if _, err := s.Ack(t.Context(), "kb", second.Token, second.Entries); err != nil {
+		t.Fatal(err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 0, 0)
+	assertSQLQueueCounts(t, s, "other-kb", 1, 0)
+	row, err := s.loadRow(t.Context(), "kb")
+	if err != nil || row.State != DatasetStateCompleted || row.LastCompletedAt == nil || row.ClaimToken != "" {
+		t.Fatalf("completed row=%+v err=%v", row, err)
+	}
+}
+
+func TestSQLSchedulerCancellationAndExpiredLeaseRecovery(t *testing.T) {
+	s := newSQLTestScheduler(t)
+	if err := s.Publish(t.Context(), "tenant", "kb", "first", "doc_completed", []string{"structure"}, []string{"Graph"}); err != nil {
+		t.Fatal(err)
+	}
+	first, ok, err := s.TryClaim(t.Context())
+	if err != nil || !ok {
+		t.Fatalf("claim ok=%v err=%v", ok, err)
+	}
+	if err := s.Publish(t.Context(), "tenant", "kb", "new", "doc_deleted", []string{"wiki"}, []string{"Wiki"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CancelInflight(t.Context(), "kb", first.Token); err != nil {
+		t.Fatal(err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 2, 0)
+	second, ok, err := s.TryClaim(t.Context())
+	if err != nil || !ok || len(second.Entries) != 2 || second.Entries[0].DocID != "new" || second.Entries[1].DocID != "first" {
+		t.Fatalf("cancel must append inflight after backlog: claim=%+v ok=%v err=%v", second, ok, err)
+	}
+	if err := s.db.Model(&entity.KnowledgeCompileDataset{}).Where("dataset_id = ?", "kb").
+		Update("claim_expires_at", time.Now().Add(-time.Minute)).Error; err != nil {
+		t.Fatal(err)
+	}
+	third, ok, err := s.TryClaim(t.Context())
+	if err != nil || !ok || third.Token == second.Token || !reflect.DeepEqual(third.Entries, second.Entries) {
+		t.Fatalf("reclaim=%+v previous=%+v ok=%v err=%v", third, second, ok, err)
+	}
+	if _, err := s.Ack(t.Context(), "kb", second.Token, second.Entries); !errors.Is(err, ErrClaimTokenMismatch) {
+		t.Fatalf("expired worker ack error=%v", err)
+	}
+	if _, err := s.Ack(t.Context(), "kb", third.Token, third.Entries); err != nil {
+		t.Fatal(err)
+	}
+	assertSQLQueueCounts(t, s, "kb", 0, 0)
+}
+
+func TestSQLSchedulerClaimRollsBackQueueMoveOnFailure(t *testing.T) {
+	s := newSQLTestScheduler(t)
+	if err := s.Publish(t.Context(), "tenant", "kb", "doc", "doc_completed", nil, nil); err != nil {
+		t.Fatal(err)
+	}
+	failure := errors.New("queue delete failed")
+	if err := s.db.Callback().Delete().Before("gorm:delete").Register("fail_backlog_delete", func(tx *gorm.DB) {
+		if tx.Statement.Table == (entity.KnowledgeCompileBacklog{}).TableName() {
+			tx.AddError(failure)
+		}
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := s.Claim(t.Context(), "kb"); !errors.Is(err, failure) {
+		t.Fatalf("claim error=%v want=%v", err, failure)
+	}
+	assertSQLQueueCounts(t, s, "kb", 1, 0)
+	row, err := s.loadRow(t.Context(), "kb")
+	if err != nil || row.ClaimToken != "" || row.State != DatasetStatePending {
+		t.Fatalf("failed claim changed lease: row=%+v err=%v", row, err)
+	}
+}

internal/ingestion/knowledge_compile/storage.md

@@ -1,5 +1,33 @@
 # Knowledge compilation storage
 
+## Scheduling queues
+
+`knowledge_compile_docs` stores one scheduling row per dataset, including its
+lease, lifecycle state, progress, and diagnostics. Events live in two tables:
+
+- `knowledge_compile_docs_backlog`: events waiting to be claimed.
+- `knowledge_compile_docs_inflight`: events claimed by a worker.
+
+Both tables contain `dataset_id`, `doc_id`, `queue_position`, `event_type`,
+`variants`, and `task_types`. They have no `id`, primary key, or unique
+constraint. Duplicate events are retained. A non-unique index on
+`(dataset_id, queue_position)` supports ordered batch reads. Routing metadata
+is stored as JSON arrays in `variants` and `task_types`.
+
+Publish, claim, acknowledgment, and recovery lock the parent dataset row and
+change queue rows in the same transaction. Positions preserve append order;
+recovery appends inflight events after the existing backlog. The dataset lease
+and claim token continue to control worker ownership. NATS only wakes workers.
+
+For an existing database, stop all services that publish or consume compilation
+events, then run `bin/ragflow_server --migrate` before restarting them. The
+migration copies each dataset's JSON queues into ordered rows, retains
+duplicates and routing metadata, and drops `backlog_doc_ids` and
+`inflight_doc_ids`. An interrupted migration can be rerun while services remain
+stopped. Server startup rejects a database that still has either source column.
+
+## Compiled products
+
 `compile_kwd` identifies the producing knowledge compilation kind. New rows use
 `graph`, `mind_map`, `page_index`, `tree`, `timeline`, or `wiki`. The existing
 session kinds remain `session_graph` and `session_essence`.

internal/ingestion/service/ingestion_service.go

@@ -101,7 +101,7 @@ type Ingestor struct {
 	// Option E) owned by this ingestor. It is driven by kcConcurrency owned
 	// worker goroutines, each running the Consumer's Run loop (poll MySQL
 	// scheduling rows + NATS notify → claim closed batch → merge → ack). The
-	// MySQL knowledge_compile_docs table — not the broker — is the scheduling
+	// MySQL knowledge_compile_docs and its queue tables are the scheduling
 	// system of record and the source of same-KB serialization, so different
 	// datasets compile in parallel while the same dataset is serialized by its
 	// claim row. The workers are started/joined within the ingestor (via
@@ -413,7 +413,7 @@ func (e *Ingestor) SetKnowledgeCompileConcurrency(n int32) {
 }
 
 // startDatasetKnowledgeCompile provisions the dataset-level compile scheduling
-// store (MySQL knowledge_compile_docs + NATS notify subject), builds the
+// store (MySQL scheduling tables + NATS notify subject), builds the
 // Consumer, and starts the owned compile-worker pool. The consumer is driven by
 // the ingestor's own goroutine set — kcConcurrency compileLoop workers, each
 // running the Consumer's Run loop (see §11.7 of knowledge_compile_design.md) —

internal/service/dataset/compilation_status.go

@@ -2,7 +2,6 @@ package dataset
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"strings"
 	"time"
@@ -57,76 +56,74 @@ func (d *DatasetService) GetDatasetCompilationStatus(ctx context.Context, userID
 	if db == nil {
 		return st, common.CodeSuccess, nil
 	}
-	var row entity.KnowledgeCompileDataset
-	err := db.WithContext(ctx).
-		Where("dataset_id = ?", datasetID).
-		First(&row).Error
-	if errors.Is(err, gorm.ErrRecordNotFound) {
-		return st, common.CodeSuccess, nil // never compiled -> idle
-	}
-	if err != nil {
-		return st, common.CodeServerError, err
-	}
-	st.State = row.State
-	if st.State == "" {
-		st.State = entity.DatasetStateIdle
-	}
-	st.Error = row.ErrorMsg
-	st.Progress = row.Progress
-	st.CurrentPhase = row.CurrentPhase
-	st.ProgressMsg = row.ProgressMsg
-	if requestedKind != "" {
-		inflight, backlog := compilationStatusCounts(row.InflightDocIDs, row.BacklogDocIDs, normalizedKind)
-		if inflight == 0 && backlog == 0 && (row.InflightDocIDs != "" || row.BacklogDocIDs != "") {
+	err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var row entity.KnowledgeCompileDataset
+		if err := tx.Where("dataset_id = ?", datasetID).First(&row).Error; err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				return nil // never compiled -> idle
+			}
+			return err
+		}
+		st.State = row.State
+		if st.State == "" {
+			st.State = entity.DatasetStateIdle
+		}
+		st.Error = row.ErrorMsg
+		st.Progress = row.Progress
+		st.CurrentPhase = row.CurrentPhase
+		st.ProgressMsg = row.ProgressMsg
+		var err error
+		st.Inflight, err = countCompilationStatusEntries(tx, (entity.KnowledgeCompileInflight{}).TableName(), datasetID, normalizedKind)
+		if err != nil {
+			return err
+		}
+		st.Backlog, err = countCompilationStatusEntries(tx, (entity.KnowledgeCompileBacklog{}).TableName(), datasetID, normalizedKind)
+		if err != nil {
+			return err
+		}
+		if requestedKind != "" && st.Inflight == 0 && st.Backlog == 0 {
 			st.State = entity.DatasetStateIdle
 			st.Error = ""
 			st.Progress = 0
 			st.CurrentPhase = ""
 			st.ProgressMsg = ""
 		}
-		st.Inflight = inflight
-		st.Backlog = backlog
-	} else {
-		st.Inflight = jsonArrayLen(row.InflightDocIDs)
-		st.Backlog = jsonArrayLen(row.BacklogDocIDs)
+		st.LastCompletedAt = row.LastCompletedAt
+		st.UpdatedAt = row.UpdatedAt
+		return nil
+	})
+	if err != nil {
+		return CompilationStatus{}, common.CodeServerError, err
 	}
-	st.LastCompletedAt = row.LastCompletedAt
-	st.UpdatedAt = row.UpdatedAt
 	return st, common.CodeSuccess, nil
 }
 
-type compilationStatusEntry struct {
-	Variants  []string `json:"variants"`
-	TaskTypes []string `json:"task_types"`
-}
-
-func compilationStatusCounts(inflightJSON, backlogJSON, kind string) (int, int) {
-	return countCompilationStatusEntries(inflightJSON, kind), countCompilationStatusEntries(backlogJSON, kind)
-}
-
-func countCompilationStatusEntries(raw, kind string) int {
-	entries := parseCompilationStatusEntries(raw)
+func countCompilationStatusEntries(db *gorm.DB, table, datasetID, kind string) (int, error) {
+	query := db.Table(table).Where("dataset_id = ?", datasetID)
+	if kind == "" {
+		var count int64
+		err := query.Count(&count).Error
+		return int(count), err
+	}
+	rows, err := query.Select("variants", "task_types").Rows()
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
 	count := 0
-	for _, entry := range entries {
+	for rows.Next() {
+		var entry entity.KnowledgeCompileEntry
+		if err := db.ScanRows(rows, &entry); err != nil {
+			return 0, err
+		}
 		if compilationStatusEntryMatches(entry, kind) {
 			count++
 		}
 	}
-	return count
+	return count, rows.Err()
 }
 
-func parseCompilationStatusEntries(raw string) []compilationStatusEntry {
-	if raw == "" {
-		return nil
-	}
-	var entries []compilationStatusEntry
-	if err := json.Unmarshal([]byte(raw), &entries); err != nil {
-		return nil
-	}
-	return entries
-}
-
-func compilationStatusEntryMatches(entry compilationStatusEntry, kind string) bool {
+func compilationStatusEntryMatches(entry entity.KnowledgeCompileEntry, kind string) bool {
 	for _, taskType := range entry.TaskTypes {
 		if normalizeCompilationKind(taskType) == kind {
 			return true
@@ -169,18 +166,3 @@ func normalizeCompilationKind(kind string) string {
 		return ""
 	}
 }
-
-// jsonArrayLen counts the top-level elements of a JSON array stored as TEXT.
-// The *_doc_ids columns hold a `[]BacklogEntry` array ({doc_id,event_type,seq}),
-// so each element is one scheduling entry (NOT a deduplicated doc). Empty or
-// malformed strings count as 0.
-func jsonArrayLen(s string) int {
-	if s == "" {
-		return 0
-	}
-	var arr []json.RawMessage
-	if err := json.Unmarshal([]byte(s), &arr); err != nil {
-		return 0
-	}
-	return len(arr)
-}

internal/service/dataset/compilation_status_test.go

@@ -26,43 +26,48 @@ import (
 	"ragflow/internal/entity"
 )
 
-// TestJSONArrayLen locks the inflight/backlog count derivation (plan v4.1
-// §9.3): counts are BacklogEntry array lengths, not deduplicated doc counts.
-func TestJSONArrayLen(t *testing.T) {
-	cases := []struct {
-		name string
-		in   string
-		want int
-	}{
-		{name: "empty", in: "", want: 0},
-		{name: "empty array", in: "[]", want: 0},
-		{name: "single entry", in: `[{"doc_id":"d1","event_type":"completed","seq":1}]`, want: 1},
-		{name: "two entries same doc", in: `[{"doc_id":"d1","event_type":"completed","seq":1},{"doc_id":"d1","event_type":"deleted","seq":2}]`, want: 2},
-		{name: "malformed", in: "{not json", want: 0},
-		{name: "null", in: "null", want: 0},
-	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			if got := jsonArrayLen(tc.in); got != tc.want {
-				t.Fatalf("jsonArrayLen(%q) = %d, want %d", tc.in, got, tc.want)
-			}
-		})
-	}
-}
-
 // setupCompilationStatusTestDB migrates the minimal schema for
 // GetDatasetCompilationStatus (Knowledgebase for the Accessible check plus the
 // KnowledgeCompileDataset scheduling row) and pushes it onto dao.DB.
 func setupCompilationStatusTestDB(t *testing.T) *gorm.DB {
 	t.Helper()
 	db := setupServiceTestDB(t)
-	if err := db.AutoMigrate(&entity.KnowledgeCompileDataset{}); err != nil {
+	if err := db.AutoMigrate(&entity.KnowledgeCompileDataset{}, &entity.KnowledgeCompileBacklog{}, &entity.KnowledgeCompileInflight{}); err != nil {
 		t.Fatalf("migrate knowledge_compile_docs: %v", err)
 	}
 	pushServiceDB(t, db)
 	return db
 }
 
+func TestCompilationQueueCountsPreserveDuplicatesAndRouting(t *testing.T) {
+	db := setupCompilationStatusTestDB(t)
+	entries := []entity.KnowledgeCompileEntry{
+		{DatasetID: "kb", DocID: "same", QueuePosition: 1, TaskTypes: []string{"Graph"}},
+		{DatasetID: "kb", DocID: "same", QueuePosition: 2, TaskTypes: []string{"Graph"}},
+		{DatasetID: "kb", DocID: "timeline", QueuePosition: 3, Variants: []string{"structure"}, TaskTypes: []string{"Timeline"}},
+		{DatasetID: "kb", DocID: "structure", QueuePosition: 4, Variants: []string{"structure"}},
+		{DatasetID: "kb", DocID: "wiki", QueuePosition: 5, Variants: []string{"wiki"}},
+		{DatasetID: "kb", DocID: "unscoped", QueuePosition: 6},
+		{DatasetID: "other-kb", DocID: "same", QueuePosition: 1, TaskTypes: []string{"Graph"}},
+	}
+	for _, table := range []string{(entity.KnowledgeCompileBacklog{}).TableName(), (entity.KnowledgeCompileInflight{}).TableName()} {
+		if err := db.Table(table).Create(&entries).Error; err != nil {
+			t.Fatal(err)
+		}
+		for _, tc := range []struct {
+			kind string
+			want int
+		}{
+			{"", 6}, {"Graph", 4}, {"Timeline", 3}, {"PageIndex", 2}, {"Wiki", 2},
+		} {
+			count, err := countCompilationStatusEntries(db, table, "kb", tc.kind)
+			if err != nil || count != tc.want {
+				t.Fatalf("%s kind=%q count=%d want=%d err=%v", table, tc.kind, count, tc.want, err)
+			}
+		}
+	}
+}
+
 // insertCompilationOwnerKB inserts a valid KB owned by userID (TenantID ==
 // userID, so Accessible returns true).
 func insertCompilationOwnerKB(t *testing.T, kbID, userID string) {
@@ -128,8 +133,6 @@ func TestGetDatasetCompilationStatus_FullOutput(t *testing.T) {
 	row := entity.KnowledgeCompileDataset{
 		DatasetID:       "kb-full",
 		TenantID:        "user-1",
-		BacklogDocIDs:   `[{"doc_id":"d3","event_type":"completed","task_types":["Timeline"]}]`,
-		InflightDocIDs:  `[{"doc_id":"d1","event_type":"completed","task_types":["Graph"]},{"doc_id":"d2","event_type":"completed","task_types":["Graph"]}]`,
 		State:           entity.DatasetStateRunning,
 		ErrorMsg:        "merge failed: boom",
 		Progress:        0.72,
@@ -140,6 +143,17 @@ func TestGetDatasetCompilationStatus_FullOutput(t *testing.T) {
 	if err := db.Create(&row).Error; err != nil {
 		t.Fatalf("insert scheduling row: %v", err)
 	}
+	if err := db.Create(&entity.KnowledgeCompileBacklog{KnowledgeCompileEntry: entity.KnowledgeCompileEntry{
+		DatasetID: row.DatasetID, DocID: "d3", QueuePosition: 1, EventType: "completed", TaskTypes: []string{"Timeline"},
+	}}).Error; err != nil {
+		t.Fatalf("insert backlog: %v", err)
+	}
+	if err := db.Create(&[]entity.KnowledgeCompileInflight{
+		{KnowledgeCompileEntry: entity.KnowledgeCompileEntry{DatasetID: row.DatasetID, DocID: "d1", QueuePosition: 1, EventType: "completed", TaskTypes: []string{"Graph"}}},
+		{KnowledgeCompileEntry: entity.KnowledgeCompileEntry{DatasetID: row.DatasetID, DocID: "d2", QueuePosition: 2, EventType: "completed", TaskTypes: []string{"Graph"}}},
+	}).Error; err != nil {
+		t.Fatalf("insert inflight: %v", err)
+	}
 
 	st, code, err := testCompilationStatusService().GetDatasetCompilationStatus(
 		t.Context(), "user-1", "kb-full", "graph")
@@ -176,11 +190,9 @@ func TestGetDatasetCompilationStatus_Unauthorized(t *testing.T) {
 	// Owner is user-1; a different user-2 must be denied.
 	insertCompilationOwnerKB(t, "kb-other", "user-1")
 	if err := dao.DB.Create(&entity.KnowledgeCompileDataset{
-		DatasetID:      "kb-other",
-		TenantID:       "user-1",
-		BacklogDocIDs:  "[]",
-		InflightDocIDs: "[]",
-		State:          entity.DatasetStatePending,
+		DatasetID: "kb-other",
+		TenantID:  "user-1",
+		State:     entity.DatasetStatePending,
 	}).Error; err != nil {
 		t.Fatalf("insert scheduling row: %v", err)
 	}