Enforce one log entry per line with a unique index (#7004)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

server/model/log.go

@@ -27,9 +27,9 @@ const (
 
 type LogEntry struct {
 	ID      int64        `json:"id"       xorm:"pk autoincr 'id'"`
-	StepID  int64        `json:"step_id"  xorm:"INDEX 'step_id'"`
+	StepID  int64        `json:"step_id"  xorm:"UNIQUE(s) INDEX 'step_id'"`
 	Time    int64        `json:"time"     xorm:"'time'"`
-	Line    int          `json:"line"     xorm:"'line'"`
+	Line    int          `json:"line"     xorm:"UNIQUE(s) 'line'"`
 	Data    []byte       `json:"data"     xorm:"LONGBLOB"`
 	Created int64        `json:"-"        xorm:"created"`
 	Type    LogEntryType `json:"type"     xorm:"'type'"`

server/store/datastore/log_test.go

@@ -129,3 +129,35 @@ func TestLogFindOrdersByLine(t *testing.T) {
 	assert.Equal(t, []int{0, 1, 2, 3}, lines)
 	assert.Equal(t, []string{"first", "second", "third", "fourth"}, data)
 }
+
+func TestLogAppendRejectsResentEntries(t *testing.T) {
+	store, closer := newTestStore(t, new(model.Step), new(model.LogEntry))
+	defer closer()
+
+	step := model.Step{
+		ID: 1,
+	}
+
+	assert.NoError(t, store.LogAppend(&step, []*model.LogEntry{
+		{StepID: step.ID, Data: []byte("hello"), Line: 0, Time: 0},
+		{StepID: step.ID, Data: []byte("world"), Line: 1, Time: 10},
+	}))
+	// the agent retries with the batch it already sent
+	assert.Error(t, store.LogAppend(&step, []*model.LogEntry{
+		{StepID: step.ID, Data: []byte("hello"), Line: 0, Time: 0},
+		{StepID: step.ID, Data: []byte("world"), Line: 1, Time: 10},
+	}))
+
+	logEntries, err := store.LogFind(&step)
+	assert.NoError(t, err)
+
+	lines := make([]int, 0, len(logEntries))
+	data := make([]string, 0, len(logEntries))
+	for _, logEntry := range logEntries {
+		lines = append(lines, logEntry.Line)
+		data = append(data, string(logEntry.Data))
+	}
+
+	assert.Equal(t, []int{0, 1}, lines)
+	assert.Equal(t, []string{"hello", "world"}, data)
+}

server/store/datastore/migration/030_deduplicate_log_entries.go

@@ -0,0 +1,52 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package migration
+
+import (
+	"src.techknowlogick.com/xormigrate"
+	"xorm.io/xorm"
+)
+
+// Clears the way for the UNIQUE(step_id, line) index on log_entries by keeping
+// the lowest id of every line number and dropping the rest.
+//
+// The inner query is restricted to line numbers that actually occur more than
+// once, so the work stays proportional to the duplicates rather than to the
+// size of the table. The extra nesting around it lets MySQL read from the
+// table it deletes from.
+//
+// This migration must not be marked Long: the UNIQUE index is created from the
+// model on every start, and skipping the cleanup would leave that failing.
+var deduplicateLogEntries = xormigrate.Migration{
+	ID: "deduplicate-log-entries",
+	MigrateSession: func(sess *xorm.Session) error {
+		_, err := sess.Exec(`DELETE FROM log_entries WHERE id IN (
+			SELECT id FROM (
+				SELECT entries.id AS id
+				FROM log_entries entries
+				JOIN (
+					SELECT step_id, line, MIN(id) AS keep_id
+					FROM log_entries
+					GROUP BY step_id, line
+					HAVING COUNT(*) > 1
+				) duplicates
+				ON duplicates.step_id = entries.step_id AND duplicates.line = entries.line
+				WHERE entries.id > duplicates.keep_id
+			) removable
+		);`)
+
+		return err
+	},
+}

server/store/datastore/migration/030_deduplicate_log_entries_test.go

@@ -0,0 +1,99 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package migration
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type logEntryV030 struct {
+	ID     int64  `xorm:"pk autoincr 'id'"`
+	StepID int64  `xorm:"'step_id'"`
+	Line   int    `xorm:"'line'"`
+	Data   []byte `xorm:"LONGBLOB"`
+}
+
+func (logEntryV030) TableName() string { return "log_entries" }
+
+func TestDeduplicateLogEntries(t *testing.T) {
+	engine, closeDB := testDB(t, true)
+	defer closeDB()
+
+	require.NoError(t, engine.Sync(new(logEntryV030)))
+
+	// Steps of this test's own: the datastore tests run against the same
+	// database and use low step ids, so sharing them would have the two
+	// fixtures collide.
+	const stepA, stepB = 30001, 30002
+
+	_, err := engine.Insert([]*logEntryV030{
+		// two entries of the first step were resent, the lowest id of each is kept
+		{ID: 301, StepID: stepA, Line: 0, Data: []byte("hello")},
+		{ID: 302, StepID: stepA, Line: 1, Data: []byte("world")},
+		{ID: 303, StepID: stepA, Line: 0, Data: []byte("hello")},
+		{ID: 304, StepID: stepA, Line: 1, Data: []byte("world")},
+		// a line number repeated three times collapses to one
+		{ID: 305, StepID: stepA, Line: 2, Data: []byte("thrice")},
+		{ID: 306, StepID: stepA, Line: 2, Data: []byte("thrice")},
+		{ID: 307, StepID: stepA, Line: 2, Data: []byte("thrice")},
+		// the same line numbers under another step must survive
+		{ID: 308, StepID: stepB, Line: 0, Data: []byte("other")},
+		{ID: 309, StepID: stepB, Line: 1, Data: []byte("other")},
+	})
+	require.NoError(t, err)
+	defer func() {
+		_, _ = engine.Exec("DELETE FROM log_entries WHERE step_id IN (?, ?);", stepA, stepB)
+	}()
+
+	sess := engine.NewSession()
+	defer sess.Close()
+	require.NoError(t, deduplicateLogEntries.MigrateSession(sess))
+	require.NoError(t, sess.Commit())
+
+	var remaining []*logEntryV030
+	require.NoError(t, engine.In("step_id", stepA, stepB).Asc("id").Find(&remaining))
+
+	ids := make([]int64, 0, len(remaining))
+	for _, entry := range remaining {
+		ids = append(ids, entry.ID)
+	}
+
+	// the lowest id of every (step_id, line) pair, and nothing else
+	assert.Equal(t, []int64{301, 302, 305, 308, 309}, ids)
+}
+
+// TestMigrateRemovesDuplicateLogEntries runs the full migration over an
+// existing database carrying resent log entries. The cleanup has to happen
+// before the UNIQUE(step_id, line) index is created from the model, otherwise
+// creating that index fails and the server does not start.
+func TestMigrateRemovesDuplicateLogEntries(t *testing.T) {
+	engine, closeDB := testDB(t, false)
+	defer closeDB()
+
+	// the fixture already holds (step_id 2, line 0); store it a second time
+	_, err := engine.Exec(
+		"INSERT INTO log_entries (id, step_id, time, line, data, created, type) VALUES (?,?,?,?,?,?,?)",
+		900001, 2, 0, 0, []byte("resent"), 1641630525, 0)
+	require.NoError(t, err)
+
+	require.NoError(t, Migrate(t.Context(), engine, true))
+
+	res, err := engine.QueryString("SELECT COUNT(*) AS total FROM log_entries WHERE step_id = 2 AND line = 0")
+	require.NoError(t, err)
+	assert.Equal(t, "1", res[0]["total"], "the resent entry should have been removed")
+}

server/store/datastore/migration/migration.go

@@ -58,6 +58,7 @@ var migrationTasks = []*xormigrate.Migration{
 	&addCronField,
 	&updatePipelineStructureTagsReleases,
 	&replaceZeroForgeIDsInUsers,
+	&deduplicateLogEntries,
 }
 
 var allBeans = []any{