#11 fix(kvstore): store CI job contexts under their own key

state: merged · 6494d11 · fast-forward
branch: fix/ci-job-contexts-key → main
opened: bnema · 2d ago
updated: 2d ago

Fixes the KV log growth reported in ermite-kv#1.

Cause

Each CI job record carried its contexts JSON (~24 KB, mostly the github.event payload). Every step update rewrote the whole record: about 58 rewrites per job and about 2 MB of log per CI run. In production, job/ keys made up 97% of the log bytes.

Change

Contexts are stored under jobctx/<run>/<attempt>/<id>, in the same batch as the job, and only when missing or changed. A step update now rewrites about 1 KB.
Every job read path hydrates contexts from the same snapshot.
Run cleanup (sweep, repo deletion) removes jobctx/ keys for all attempts.
Records that still hold contexts_json inline are read through a fallback, and their next write moves the value to jobctx/.

Checks

go build, go vet and go test ./... pass; -race passes on kvstore.
Review: no findings.

Commit	Subject	Author	Date	Signature	Status
6494d11	refactor(kvstore): list job contexts keys in their own helper	brice	2d ago	GPG Verified
e908c5b	fix(kvstore): store CI job contexts under their own key	brice	2d ago	GPG Verified

Checks 8

CI · pull_request
Image success
Lint success
Protobuf and generated code success
Test success
CI · push
Image success
Lint success
Protobuf and generated code success
Test success

4 files · +158 -8

8/8

M gen/ermite/storage/ci.pb.go +2 -1


@@ -978,7 +978,8 @@ type CIJob struct {
978	978		// keeps its own attempt.
979	979		RunAttempt int32 `protobuf:"varint,38,opt,name=run_attempt,json=runAttempt,proto3" json:"run_attempt,omitempty"`
980	980		// The contexts JSON given to the runner, computed when the job is planned
981		-	// (needs outputs are final then).
	981	+	// (needs outputs are final then). No longer written: stored under jobctx/
	982	+	// instead. Retained as a read fallback for legacy records.
982	983		ContextsJson string `protobuf:"bytes,39,opt,name=contexts_json,json=contextsJson,proto3" json:"contexts_json,omitempty"`
983	984		// The job's continue-on-error, evaluated when the job is planned: a failure
984	985		// of the job does not fail the run, does not cancel the siblings of its

 	// keeps its own attempt.
 	RunAttempt int32 `protobuf:"varint,38,opt,name=run_attempt,json=runAttempt,proto3" json:"run_attempt,omitempty"`
 	// The contexts JSON given to the runner, computed when the job is planned
-	// (needs outputs are final then).
+	// (needs outputs are final then). No longer written: stored under jobctx/
+	// instead. Retained as a read fallback for legacy records.
 	ContextsJson string `protobuf:"bytes,39,opt,name=contexts_json,json=contextsJson,proto3" json:"contexts_json,omitempty"`
 	// The job's continue-on-error, evaluated when the job is planned: a failure
 	// of the job does not fail the run, does not cancel the siblings of its

M internal/adapters/kvstore/ci.go +43 -6


@@ -26,6 +26,7 @@ import (
26	26		// run/<repo>/<id:020d> CIRun
27	27		// runsrc/<id:020d> CIRunSource
28	28		// job/<run:020d>/<attempt:010d>/<id> CIJob
	29	+	// jobctx/<run:020d>/<attempt:010d>/<id> contexts JSON
29	30		// check/<repo>/<sha>/<check id> CICheck
30	31		// runseq Sequence: last run id
31	32		// runnum/<repo>/<workflow path> Sequence: last run number

@@ -46,6 +47,7 @@ const (
46	47		runPrefix = "run/"
47	48		runSrcPrefix = "runsrc/"
48	49		jobPrefix = "job/"
	50	+	jobCtxPrefix = "jobctx/"
49	51		checkPrefix = "check/"
50	52		runSeqKey = "runseq"
51	53		runNumPrefix = "runnum/"

@@ -85,6 +87,10 @@ func jobRef(runID int64, attempt int32, id string) string {
85	87		return jobAttemptPrefix(runID, attempt) + id
86	88		}
87	89		func jobKey(runID int64, attempt int32, id string) []byte { return []byte(jobRef(runID, attempt, id)) }
	90	+	func jobContextsKey(runID int64, attempt int32, id string) []byte {
	91	+	return []byte(jobCtxPrefix + strings.TrimPrefix(jobRef(runID, attempt, id), jobPrefix))
	92	+	}
	93	+
88	94		func checksPrefix(repo domain.RepoName, sha string) string {
89	95		return checkPrefix + repo.String() + "/" + sha + "/"
90	96		}

@@ -335,7 +341,7 @@ func encodeJob(j domain.CIJob) ([]byte, error) {
335	341		Labels: j.Labels, Needs: j.Needs, RunnerId: j.RunnerID, RunnerName: j.RunnerName,
336	342		Isolation: encodeIsolation(j.Isolation), Steps: steps,
337	343		CreatedAt: encodeTime(j.CreatedAt), StartedAt: encodeTime(j.StartedAt), CompletedAt: encodeTime(j.CompletedAt),
338		-	Index: int32(j.Index), Combo: int32(j.Combo), MatrixJson: j.MatrixJSON, ContextsJson: j.ContextsJSON, ComboTotal: int32(j.ComboTotal), //nolint:gosec // bounded by the 256 job limit
	344	+	Index: int32(j.Index), Combo: int32(j.Combo), MatrixJson: j.MatrixJSON, ComboTotal: int32(j.ComboTotal), //nolint:gosec // bounded by the 256 job limit
339	345		TimeoutMinutes: j.TimeoutMinutes, FailFast: j.FailFast, ContinueOnError: j.ContinueOnError, MaxParallel: int32(j.MaxParallel), //nolint:gosec // validated at planning
340	346		Outputs: j.Outputs, ReadyAt: encodeTime(j.ReadyAt), AssignedAt: encodeTime(j.AssignedAt),
341	347		AckedAt: encodeTime(j.AckedAt), AssignCancels: int32(j.AssignCancels), //nolint:gosec // at most 3

@@ -385,6 +391,19 @@ func decodeJob(item kv.Item) (domain.CIJob, error) {
385	391		return j, nil
386	392		}
387	393
	394	+	// decodeJobWithContexts reads both parts from the same snapshot. The inline
	395	+	// field is retained only as a fallback for records written before the split.
	396	+	func decodeJobWithContexts(ctx context.Context, snap *kv.Snapshot, item kv.Item) (domain.CIJob, error) {
	397	+	j, err := decodeJob(item)
	398	+	if err != nil {
	399	+	return domain.CIJob{}, err
	400	+	}
	401	+	if contexts, ok := snap.Get(ctx, jobContextsKey(j.RunID, j.RunAttempt, j.ID)); ok {
	402	+	j.ContextsJSON = string(contexts.Value)
	403	+	}
	404	+	return j, nil
	405	+	}
	406	+
388	407		func encodeCheck(c domain.CICheck) []byte {
389	408		return mustMarshal(&storagepb.CICheck{
390	409		Id: c.ID, Repo: c.Repo.String(), Sha: c.Sha, Name: c.Name, Status: encodeStatus(c.Status),

@@ -515,6 +534,12 @@ func (c CI) stageRecords(ctx context.Context, b kv.Batch, w ports.CIWrite) (gu
515	534		old = &o
516	535		}
517	536		}
	537	+	// Compare the separate value, not the legacy inline field: an old
	538	+	// record must materialize contexts before encodeJob drops that field.
	539	+	contextsKey := jobContextsKey(j.RunID, j.RunAttempt, j.ID)
	540	+	if contexts, ok := c.db.Get(ctx, contextsKey); !ok \|\| string(contexts.Value) != j.ContextsJSON {
	541	+	b.Put(contextsKey, []byte(j.ContextsJSON))
	542	+	}
518	543		stage(b, key, val, j.Version)
519	544		c.jobIndexes(b, j, old)
520	545		if j.Version == 0 {

@@ -746,19 +771,21 @@ func (c *CI) ListOpenRuns(ctx context.Context) ([]domain.CIRun, error) {
746	771
747	772		// GetJob reads one job record.
748	773		func (c *CI) GetJob(ctx context.Context, runID int64, runAttempt int32, jobID string) (domain.CIJob, error) {
749		-	item, ok := c.db.Get(ctx, jobKey(runID, runAttempt, jobID))
	774	+	snap := c.db.Snapshot(ctx)
	775	+	item, ok := snap.Get(ctx, jobKey(runID, runAttempt, jobID))
750	776		if !ok {
751	777		return domain.CIJob{}, domain.ErrNotFound
752	778		}
753		-	return decodeJob(item)
	779	+	return decodeJobWithContexts(ctx, snap, item)
754	780		}
755	781
756	782		// ListJobs returns the jobs of a run at an attempt, ordered by index then
757	783		// combination.
758	784		func (c *CI) ListJobs(ctx context.Context, runID int64, runAttempt int32) ([]domain.CIJob, error) {
759	785		var out []domain.CIJob
760		-	for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(jobAttemptPrefix(runID, runAttempt))) {
761		-	j, err := decodeJob(item)
	786	+	snap := c.db.Snapshot(ctx)
	787	+	for key, item := range snap.Prefix(ctx, []byte(jobAttemptPrefix(runID, runAttempt))) {
	788	+	j, err := decodeJobWithContexts(ctx, snap, item)
762	789		if err != nil {
763	790		return nil, fmt.Errorf("decode %s: %w", key, err)
764	791		}

@@ -778,7 +805,7 @@ func (c CI) jobByRef(snap kv.Snapshot, ctx context.Context, ref string) (domai
778	805		if !ok {
779	806		return domain.CIJob{}, domain.ErrNotFound
780	807		}
781		-	return decodeJob(item)
	808	+	return decodeJobWithContexts(ctx, snap, item)
782	809		}
783	810
784	811		// FindJobByToken resolves the token index, then reads the job.

@@ -862,11 +889,21 @@ func (c *CI) ListChecks(ctx context.Context, repo domain.RepoName, sha string, l
862	889
863	890		// --- retention and deletion ---
864	891
	892	+	// jobCtxKeys lists the contexts keys of every attempt of a run.
	893	+	func jobCtxKeys(ctx context.Context, snap *kv.Snapshot, runID int64) [][]byte {
	894	+	var keys [][]byte
	895	+	for key := range snap.Prefix(ctx, []byte(jobCtxPrefix+fmt.Sprintf(idFormat, runID)+"/")) {
	896	+	keys = append(keys, slices.Clone(key))
	897	+	}
	898	+	return keys
	899	+	}
	900	+
865	901		// runKeys lists every key of a run except its record: source, jobs, indexes.
866	902		func (c CI) runKeys(ctx context.Context, snap kv.Snapshot, repo domain.RepoName, runID int64) [][]byte {
867	903		keys := [][]byte{runKey(repo, runID), runSrcKey(runID), openKey(runID)}
868	904		keys = append(keys, repoTokenKeys(ctx, snap, runID)...)
869	905		keys = append(keys, annotationKeys(ctx, snap, runID)...)
	906	+	keys = append(keys, jobCtxKeys(ctx, snap, runID)...)
870	907		for key, item := range snap.Prefix(ctx, []byte(jobsPrefix(runID))) {
871	908		keys = append(keys, slices.Clone(key))
872	909		if j, err := decodeJob(item); err == nil {

 //	run/<repo>/<id:020d>                 CIRun
 //	runsrc/<id:020d>                     CIRunSource
 //	job/<run:020d>/<attempt:010d>/<id>   CIJob
+//	jobctx/<run:020d>/<attempt:010d>/<id> contexts JSON
 //	check/<repo>/<sha>/<check id>        CICheck
 //	runseq                               Sequence: last run id
 //	runnum/<repo>/<workflow path>        Sequence: last run number
 	runPrefix     = "run/"
 	runSrcPrefix  = "runsrc/"
 	jobPrefix     = "job/"
+	jobCtxPrefix  = "jobctx/"
 	checkPrefix   = "check/"
 	runSeqKey     = "runseq"
 	runNumPrefix  = "runnum/"
 	return jobAttemptPrefix(runID, attempt) + id
 }
 func jobKey(runID int64, attempt int32, id string) []byte { return []byte(jobRef(runID, attempt, id)) }
+func jobContextsKey(runID int64, attempt int32, id string) []byte {
+	return []byte(jobCtxPrefix + strings.TrimPrefix(jobRef(runID, attempt, id), jobPrefix))
+}
 func checksPrefix(repo domain.RepoName, sha string) string {
 	return checkPrefix + repo.String() + "/" + sha + "/"
 }
 		Labels: j.Labels, Needs: j.Needs, RunnerId: j.RunnerID, RunnerName: j.RunnerName,
 		Isolation: encodeIsolation(j.Isolation), Steps: steps,
 		CreatedAt: encodeTime(j.CreatedAt), StartedAt: encodeTime(j.StartedAt), CompletedAt: encodeTime(j.CompletedAt),
-		Index: int32(j.Index), Combo: int32(j.Combo), MatrixJson: j.MatrixJSON, ContextsJson: j.ContextsJSON, ComboTotal: int32(j.ComboTotal), //nolint:gosec // bounded by the 256 job limit
+		Index: int32(j.Index), Combo: int32(j.Combo), MatrixJson: j.MatrixJSON, ComboTotal: int32(j.ComboTotal), //nolint:gosec // bounded by the 256 job limit
 		TimeoutMinutes: j.TimeoutMinutes, FailFast: j.FailFast, ContinueOnError: j.ContinueOnError, MaxParallel: int32(j.MaxParallel), //nolint:gosec // validated at planning
 		Outputs: j.Outputs, ReadyAt: encodeTime(j.ReadyAt), AssignedAt: encodeTime(j.AssignedAt),
 		AckedAt: encodeTime(j.AckedAt), AssignCancels: int32(j.AssignCancels), //nolint:gosec // at most 3
 	return j, nil
 }
+// decodeJobWithContexts reads both parts from the same snapshot. The inline
+// field is retained only as a fallback for records written before the split.
+func decodeJobWithContexts(ctx context.Context, snap *kv.Snapshot, item kv.Item) (domain.CIJob, error) {
+	j, err := decodeJob(item)
+	if err != nil {
+		return domain.CIJob{}, err
+	}
+	if contexts, ok := snap.Get(ctx, jobContextsKey(j.RunID, j.RunAttempt, j.ID)); ok {
+		j.ContextsJSON = string(contexts.Value)
+	}
+	return j, nil
+}
 func encodeCheck(c domain.CICheck) []byte {
 	return mustMarshal(&storagepb.CICheck{
 		Id: c.ID, Repo: c.Repo.String(), Sha: c.Sha, Name: c.Name, Status: encodeStatus(c.Status),
 				old = &o
 			}
 		}
+		// Compare the separate value, not the legacy inline field: an old
+		// record must materialize contexts before encodeJob drops that field.
+		contextsKey := jobContextsKey(j.RunID, j.RunAttempt, j.ID)
+		if contexts, ok := c.db.Get(ctx, contextsKey); !ok || string(contexts.Value) != j.ContextsJSON {
+			b.Put(contextsKey, []byte(j.ContextsJSON))
+		}
 		stage(b, key, val, j.Version)
 		c.jobIndexes(b, j, old)
 		if j.Version == 0 {
 // GetJob reads one job record.
 func (c *CI) GetJob(ctx context.Context, runID int64, runAttempt int32, jobID string) (domain.CIJob, error) {
-	item, ok := c.db.Get(ctx, jobKey(runID, runAttempt, jobID))
+	snap := c.db.Snapshot(ctx)
+	item, ok := snap.Get(ctx, jobKey(runID, runAttempt, jobID))
 	if !ok {
 		return domain.CIJob{}, domain.ErrNotFound
 	}
-	return decodeJob(item)
+	return decodeJobWithContexts(ctx, snap, item)
 }
 // ListJobs returns the jobs of a run at an attempt, ordered by index then
 // combination.
 func (c *CI) ListJobs(ctx context.Context, runID int64, runAttempt int32) ([]domain.CIJob, error) {
 	var out []domain.CIJob
-	for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(jobAttemptPrefix(runID, runAttempt))) {
+	snap := c.db.Snapshot(ctx)
-		j, err := decodeJob(item)
+	for key, item := range snap.Prefix(ctx, []byte(jobAttemptPrefix(runID, runAttempt))) {
+		j, err := decodeJobWithContexts(ctx, snap, item)
 		if err != nil {
 			return nil, fmt.Errorf("decode %s: %w", key, err)
 		}
 	if !ok {
 		return domain.CIJob{}, domain.ErrNotFound
 	}
-	return decodeJob(item)
+	return decodeJobWithContexts(ctx, snap, item)
 }
 // FindJobByToken resolves the token index, then reads the job.
 // --- retention and deletion ---
+// jobCtxKeys lists the contexts keys of every attempt of a run.
+func jobCtxKeys(ctx context.Context, snap *kv.Snapshot, runID int64) [][]byte {
+	var keys [][]byte
+	for key := range snap.Prefix(ctx, []byte(jobCtxPrefix+fmt.Sprintf(idFormat, runID)+"/")) {
+		keys = append(keys, slices.Clone(key))
+	}
+	return keys
+}
 // runKeys lists every key of a run except its record: source, jobs, indexes.
 func (c *CI) runKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName, runID int64) [][]byte {
 	keys := [][]byte{runKey(repo, runID), runSrcKey(runID), openKey(runID)}
 	keys = append(keys, repoTokenKeys(ctx, snap, runID)...)
 	keys = append(keys, annotationKeys(ctx, snap, runID)...)
+	keys = append(keys, jobCtxKeys(ctx, snap, runID)...)
 	for key, item := range snap.Prefix(ctx, []byte(jobsPrefix(runID))) {
 		keys = append(keys, slices.Clone(key))
 		if j, err := decodeJob(item); err == nil {

A internal/adapters/kvstore/ci_contexts_test.go +111 -0

@@ -0,0 +1,111 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	kv "bnema.dev/ermite-kv"
	12	+	storagepb "bnema.dev/ermite/gen/ermite/storage"
	13	+	"bnema.dev/ermite/internal/domain"
	14	+	"bnema.dev/ermite/internal/ports"
	15	+	"github.com/stretchr/testify/require"
	16	+	"google.golang.org/protobuf/proto"
	17	+	)
	18	+
	19	+	func TestCIJobContextsStorage(t *testing.T) {
	20	+	ctx := context.Background()
	21	+	s, err := Open(ctx, t.TempDir())
	22	+	require.NoError(t, err)
	23	+	t.Cleanup(func() { require.NoError(t, s.Close(ctx)) })
	24	+	repo, err := domain.ParseRepoName("proj")
	25	+	require.NoError(t, err)
	26	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
	27	+	ci := s.CI()
	28	+	payload := `{"event":"` + strings.Repeat("payload", 3500) + `"}`
	29	+	j := domain.CIJob{ID: "build", Repo: repo, RunID: 1, RunAttempt: 1, Attempt: 1, Phase: domain.CIPhaseReady, ContextsJSON: payload}
	30	+	save := func() {
	31	+	v, err := ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
	32	+	require.NoError(t, err)
	33	+	j.Version = v
	34	+	}
	35	+	save()
	36	+	contextsKey := jobContextsKey(1, 1, j.ID)
	37	+	initial, ok := ci.db.Get(ctx, contextsKey)
	38	+	require.True(t, ok)
	39	+	for i := range 5 {
	40	+	j.Steps = []domain.CIStep{{Number: int32(i + 1), Name: "step", State: domain.CIStepInProgress}}
	41	+	save()
	42	+	stored, ok := ci.db.Get(ctx, jobKey(1, 1, j.ID))
	43	+	require.True(t, ok)
	44	+	require.NotContains(t, string(stored.Value), payload)
	45	+	var pb storagepb.CIJob
	46	+	require.NoError(t, proto.Unmarshal(stored.Value, &pb))
	47	+	require.Empty(t, pb.GetContextsJson())
	48	+	contexts, ok := ci.db.Get(ctx, contextsKey)
	49	+	require.True(t, ok)
	50	+	require.Equal(t, initial.Version, contexts.Version, "unchanged contexts must not be rewritten")
	51	+	got, err := ci.GetJob(ctx, 1, 1, j.ID)
	52	+	require.NoError(t, err)
	53	+	require.Equal(t, payload, got.ContextsJSON)
	54	+	jobs, err := ci.ListJobs(ctx, 1, 1)
	55	+	require.NoError(t, err)
	56	+	require.Equal(t, payload, jobs[0].ContextsJSON)
	57	+	ready, err := ci.ListReadyJobs(ctx, 0)
	58	+	require.NoError(t, err)
	59	+	require.Equal(t, payload, ready[0].ContextsJSON)
	60	+	}
	61	+	// A kept job in a re-run gets an independent contexts key.
	62	+	j.RunAttempt = 2
	63	+	j.Version = 0
	64	+	save()
	65	+	copied, ok := ci.db.Get(ctx, jobContextsKey(1, 2, j.ID))
	66	+	require.True(t, ok)
	67	+	require.Equal(t, payload, string(copied.Value))
	68	+	j.ContextsJSON = ""
	69	+	save()
	70	+	got, err := ci.GetJob(ctx, 1, 2, j.ID)
	71	+	require.NoError(t, err)
	72	+	require.Empty(t, got.ContextsJSON)
	73	+	// Sweep must remove contexts from every attempt as well as job records.
	74	+	_, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{{Repo: repo, ID: 1, Status: domain.CICompleted, CompletedAt: time.Now().Add(-time.Hour)}}})
	75	+	require.NoError(t, err)
	76	+	ids, err := ci.Sweep(ctx, time.Now())
	77	+	require.NoError(t, err)
	78	+	require.Equal(t, []int64{1}, ids)
	79	+	for range ci.db.Snapshot(ctx).Prefix(ctx, []byte(jobCtxPrefix)) {
	80	+	t.Fatal("orphan contexts key")
	81	+	}
	82	+	}
	83	+
	84	+	func TestCIJobLegacyContexts(t *testing.T) {
	85	+	ctx := context.Background()
	86	+	s, err := Open(ctx, t.TempDir())
	87	+	require.NoError(t, err)
	88	+	t.Cleanup(func() { require.NoError(t, s.Close(ctx)) })
	89	+	ci := s.CI()
	90	+	var b kv.Batch
	91	+	b.Put(jobKey(1, 1, "build"), mustMarshal(&storagepb.CIJob{Id: "build", Repo: "proj", RunId: 1, RunAttempt: 1, ContextsJson: `{"legacy":true}`}))
	92	+	_, err = ci.db.Apply(ctx, &b)
	93	+	require.NoError(t, err)
	94	+	j, err := ci.GetJob(ctx, 1, 1, "build")
	95	+	require.NoError(t, err)
	96	+	require.Equal(t, `{"legacy":true}`, j.ContextsJSON)
	97	+	jobs, err := ci.ListJobs(ctx, 1, 1)
	98	+	require.NoError(t, err)
	99	+	require.Equal(t, j.ContextsJSON, jobs[0].ContextsJSON)
	100	+	// Updating the legacy job moves its contexts out of the protobuf.
	101	+	_, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
	102	+	require.NoError(t, err)
	103	+	got, err := ci.GetJob(ctx, 1, 1, "build")
	104	+	require.NoError(t, err)
	105	+	require.Equal(t, j.ContextsJSON, got.ContextsJSON)
	106	+	item, ok := ci.db.Get(ctx, jobKey(1, 1, "build"))
	107	+	require.True(t, ok)
	108	+	pbJob, err := decodeJob(item)
	109	+	require.NoError(t, err)
	110	+	require.Empty(t, pbJob.ContextsJSON)
	111	+	}

@@ -0,0 +1,111 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	kv "bnema.dev/ermite-kv"
	12	+	storagepb "bnema.dev/ermite/gen/ermite/storage"
	13	+	"bnema.dev/ermite/internal/domain"
	14	+	"bnema.dev/ermite/internal/ports"
	15	+	"github.com/stretchr/testify/require"
	16	+	"google.golang.org/protobuf/proto"
	17	+	)
	18	+
	19	+	func TestCIJobContextsStorage(t *testing.T) {
	20	+	ctx := context.Background()
	21	+	s, err := Open(ctx, t.TempDir())
	22	+	require.NoError(t, err)
	23	+	t.Cleanup(func() { require.NoError(t, s.Close(ctx)) })
	24	+	repo, err := domain.ParseRepoName("proj")
	25	+	require.NoError(t, err)
	26	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
	27	+	ci := s.CI()
	28	+	payload := `{"event":"` + strings.Repeat("payload", 3500) + `"}`
	29	+	j := domain.CIJob{ID: "build", Repo: repo, RunID: 1, RunAttempt: 1, Attempt: 1, Phase: domain.CIPhaseReady, ContextsJSON: payload}
	30	+	save := func() {
	31	+	v, err := ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
	32	+	require.NoError(t, err)
	33	+	j.Version = v
	34	+	}
	35	+	save()
	36	+	contextsKey := jobContextsKey(1, 1, j.ID)
	37	+	initial, ok := ci.db.Get(ctx, contextsKey)
	38	+	require.True(t, ok)
	39	+	for i := range 5 {
	40	+	j.Steps = []domain.CIStep{{Number: int32(i + 1), Name: "step", State: domain.CIStepInProgress}}
	41	+	save()
	42	+	stored, ok := ci.db.Get(ctx, jobKey(1, 1, j.ID))
	43	+	require.True(t, ok)
	44	+	require.NotContains(t, string(stored.Value), payload)
	45	+	var pb storagepb.CIJob
	46	+	require.NoError(t, proto.Unmarshal(stored.Value, &pb))
	47	+	require.Empty(t, pb.GetContextsJson())
	48	+	contexts, ok := ci.db.Get(ctx, contextsKey)
	49	+	require.True(t, ok)
	50	+	require.Equal(t, initial.Version, contexts.Version, "unchanged contexts must not be rewritten")
	51	+	got, err := ci.GetJob(ctx, 1, 1, j.ID)
	52	+	require.NoError(t, err)
	53	+	require.Equal(t, payload, got.ContextsJSON)
	54	+	jobs, err := ci.ListJobs(ctx, 1, 1)
	55	+	require.NoError(t, err)
	56	+	require.Equal(t, payload, jobs[0].ContextsJSON)
	57	+	ready, err := ci.ListReadyJobs(ctx, 0)
	58	+	require.NoError(t, err)
	59	+	require.Equal(t, payload, ready[0].ContextsJSON)
	60	+	}
	61	+	// A kept job in a re-run gets an independent contexts key.
	62	+	j.RunAttempt = 2
	63	+	j.Version = 0
	64	+	save()
	65	+	copied, ok := ci.db.Get(ctx, jobContextsKey(1, 2, j.ID))
	66	+	require.True(t, ok)
	67	+	require.Equal(t, payload, string(copied.Value))
	68	+	j.ContextsJSON = ""
	69	+	save()
	70	+	got, err := ci.GetJob(ctx, 1, 2, j.ID)
	71	+	require.NoError(t, err)
	72	+	require.Empty(t, got.ContextsJSON)
	73	+	// Sweep must remove contexts from every attempt as well as job records.
	74	+	_, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{{Repo: repo, ID: 1, Status: domain.CICompleted, CompletedAt: time.Now().Add(-time.Hour)}}})
	75	+	require.NoError(t, err)
	76	+	ids, err := ci.Sweep(ctx, time.Now())
	77	+	require.NoError(t, err)
	78	+	require.Equal(t, []int64{1}, ids)
	79	+	for range ci.db.Snapshot(ctx).Prefix(ctx, []byte(jobCtxPrefix)) {
	80	+	t.Fatal("orphan contexts key")
	81	+	}
	82	+	}
	83	+
	84	+	func TestCIJobLegacyContexts(t *testing.T) {
	85	+	ctx := context.Background()
	86	+	s, err := Open(ctx, t.TempDir())
	87	+	require.NoError(t, err)
	88	+	t.Cleanup(func() { require.NoError(t, s.Close(ctx)) })
	89	+	ci := s.CI()
	90	+	var b kv.Batch
	91	+	b.Put(jobKey(1, 1, "build"), mustMarshal(&storagepb.CIJob{Id: "build", Repo: "proj", RunId: 1, RunAttempt: 1, ContextsJson: `{"legacy":true}`}))
	92	+	_, err = ci.db.Apply(ctx, &b)
	93	+	require.NoError(t, err)
	94	+	j, err := ci.GetJob(ctx, 1, 1, "build")
	95	+	require.NoError(t, err)
	96	+	require.Equal(t, `{"legacy":true}`, j.ContextsJSON)
	97	+	jobs, err := ci.ListJobs(ctx, 1, 1)
	98	+	require.NoError(t, err)
	99	+	require.Equal(t, j.ContextsJSON, jobs[0].ContextsJSON)
	100	+	// Updating the legacy job moves its contexts out of the protobuf.
	101	+	_, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
	102	+	require.NoError(t, err)
	103	+	got, err := ci.GetJob(ctx, 1, 1, "build")
	104	+	require.NoError(t, err)
	105	+	require.Equal(t, j.ContextsJSON, got.ContextsJSON)
	106	+	item, ok := ci.db.Get(ctx, jobKey(1, 1, "build"))
	107	+	require.True(t, ok)
	108	+	pbJob, err := decodeJob(item)
	109	+	require.NoError(t, err)
	110	+	require.Empty(t, pbJob.ContextsJSON)
	111	+	}

M proto/ermite/storage/ci.proto +2 -1


@@ -203,7 +203,8 @@ message CIJob {
203	203		// keeps its own attempt.
204	204		int32 run_attempt = 38;
205	205		// The contexts JSON given to the runner, computed when the job is planned
206		-	// (needs outputs are final then).
	206	+	// (needs outputs are final then). No longer written: stored under jobctx/
	207	+	// instead. Retained as a read fallback for legacy records.
207	208		string contexts_json = 39;
208	209		// The job's continue-on-error, evaluated when the job is planned: a failure
209	210		// of the job does not fail the run, does not cancel the siblings of its

   // keeps its own attempt.
   int32 run_attempt = 38;
   // The contexts JSON given to the runner, computed when the job is planned
-  // (needs outputs are final then).
+  // (needs outputs are final then). No longer written: stored under jobctx/
+  // instead. Retained as a read fallback for legacy records.
   string contexts_json = 39;
   // The job's continue-on-error, evaluated when the job is planned: a failure
   // of the job does not fail the run, does not cancel the siblings of its

Comments0 open / 0

no comments yet.

Cause

Change

Checks

Merge settings

Comments0 open / 0