diff --git a/server/model/log.go b/server/model/log.go index 640c80457..0200d7d50 100644 --- a/server/model/log.go +++ b/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'"` diff --git a/server/store/datastore/log_test.go b/server/store/datastore/log_test.go index 993130340..bb767cc2b 100644 --- a/server/store/datastore/log_test.go +++ b/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) +} diff --git a/server/store/datastore/migration/030_deduplicate_log_entries.go b/server/store/datastore/migration/030_deduplicate_log_entries.go new file mode 100644 index 000000000..3f4c3a0d2 --- /dev/null +++ b/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 + }, +} diff --git a/server/store/datastore/migration/030_deduplicate_log_entries_test.go b/server/store/datastore/migration/030_deduplicate_log_entries_test.go new file mode 100644 index 000000000..47e3566f5 --- /dev/null +++ b/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") +} diff --git a/server/store/datastore/migration/migration.go b/server/store/datastore/migration/migration.go index 0037cf5f8..68ba8dab5 100644 --- a/server/store/datastore/migration/migration.go +++ b/server/store/datastore/migration/migration.go @@ -58,6 +58,7 @@ var migrationTasks = []*xormigrate.Migration{ &addCronField, &updatePipelineStructureTagsReleases, &replaceZeroForgeIDsInUsers, + &deduplicateLogEntries, } var allBeans = []any{