Add activity history clearing
This commit is contained in:
@@ -36,6 +36,7 @@ func New(socket string, svc *service.Service, log *slog.Logger) *Server {
|
||||
mux.HandleFunc("PUT /v1/secrets/{id}", s.putSecret)
|
||||
mux.HandleFunc("DELETE /v1/secrets/{id}", s.deleteSecret)
|
||||
mux.HandleFunc("GET /v1/runs", s.runs)
|
||||
mux.HandleFunc("DELETE /v1/runs", s.clearRuns)
|
||||
mux.HandleFunc("GET /v1/filesystem/directories", s.directories)
|
||||
mux.HandleFunc("GET /v1/workloads/{kind}", s.workloads)
|
||||
mux.HandleFunc("POST /v1/jobs/{id}/run", s.runJob)
|
||||
@@ -118,6 +119,9 @@ func (s *Server) deleteSecret(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(204)
|
||||
}
|
||||
func (s *Server) runs(w http.ResponseWriter, _ *http.Request) { writeJSON(w, 200, s.service.Runs()) }
|
||||
func (s *Server) clearRuns(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, 200, map[string]int{"cleared": s.service.ClearRunHistory()})
|
||||
}
|
||||
func (s *Server) directories(w http.ResponseWriter, r *http.Request) {
|
||||
items, err := s.service.BrowseDirectories(r.URL.Query().Get("path"))
|
||||
if err != nil {
|
||||
|
||||
@@ -151,6 +151,15 @@ func (q *Queue) UpdateActive(id string, update func(*model.Run)) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (q *Queue) ClearHistory() int {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
count := len(q.history)
|
||||
q.history = nil
|
||||
q.emitLocked()
|
||||
return count
|
||||
}
|
||||
|
||||
func (q *Queue) snapshotLocked() []model.Run {
|
||||
result := append([]model.Run{}, q.history...)
|
||||
if q.active != nil {
|
||||
|
||||
@@ -28,3 +28,34 @@ func TestQueueDeduplicatesJobAndRunsTask(t *testing.T) {
|
||||
}
|
||||
q.Stop()
|
||||
}
|
||||
|
||||
func TestClearHistoryKeepsActiveAndPendingRuns(t *testing.T) {
|
||||
started := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
q := New(func(_ context.Context, run model.Run) model.Run {
|
||||
close(started)
|
||||
<-release
|
||||
run.Status = "success"
|
||||
return run
|
||||
}, nil)
|
||||
q.RestoreHistory([]model.Run{{ID: "old", Status: "success"}})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go q.Run(ctx)
|
||||
if err := q.Enqueue(model.Run{ID: "active", JobID: "active-job"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
<-started
|
||||
if err := q.Enqueue(model.Run{ID: "pending", JobID: "pending-job"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cleared := q.ClearHistory(); cleared != 1 {
|
||||
t.Fatalf("cleared = %d", cleared)
|
||||
}
|
||||
runs := q.Snapshot()
|
||||
if len(runs) != 2 || runs[0].ID != "active" || runs[1].ID != "pending" {
|
||||
t.Fatalf("runs = %#v", runs)
|
||||
}
|
||||
q.Stop()
|
||||
close(release)
|
||||
}
|
||||
|
||||
@@ -166,6 +166,7 @@ func (s *Service) EnqueueRestore(task model.RestoreTask) (model.Run, error) {
|
||||
}
|
||||
|
||||
func (s *Service) Cancel(runID string) bool { return s.queue.Cancel(runID) }
|
||||
func (s *Service) ClearRunHistory() int { return s.queue.ClearHistory() }
|
||||
|
||||
func (s *Service) Snapshots(ctx context.Context, repoID string) ([]model.Snapshot, error) {
|
||||
repo, ok := s.repository(repoID)
|
||||
|
||||
Reference in New Issue
Block a user