feat(downloads): add resume-safe pause action (#11222)

Give gallery operations distinct pause and cancel paths. Pause preserves partial download data so reinstalling the same model or backend resumes through HTTP Range, while cancel keeps its destructive semantics. Surface the action in the Activity UI and document the API behavior.

Assisted-by: Codex:gpt-5

Co-authored-by: localai-org-maint-bot <306269227+localai-org-maint-bot@users.noreply.github.com>
This commit is contained in:
localai-org-maint-botandlocalai-org-maint-bot authored and GitHub committed 2026-08-03 15:25:23 +02:00
1 parent 8f74f74b10
commit fd4ec083b9
20 files changed
+224 -43

No files matched your search

@@ -43,6 +43,40 @@ test('lists live operations and cancels one from a labelled button', async ({ pa
expect(cancelledPath).toBe('/api/operations/job-gemma/cancel') expect(cancelledPath).toBe('/api/operations/job-gemma/cancel')
}) })
test('pauses a model download without invoking destructive cancel', async ({ page }) => {
await stub(page, {
operations: [{
id: 'gemma-3-27b-it',
name: 'gemma-3-27b-it',
jobID: 'job-gemma',
progress: 22,
taskType: 'installation',
isBackend: false,
isQueued: false,
isDeletion: false,
cancellable: true,
phase: 'downloading',
}],
})
const requests = []
await page.route('**/api/operations/job-gemma/pause', (route) => {
requests.push(new URL(route.request().url()).pathname)
return route.fulfill({ contentType: 'application/json', body: '{}' })
})
await page.route('**/api/operations/job-gemma/cancel', (route) => {
requests.push(new URL(route.request().url()).pathname)
return route.fulfill({ contentType: 'application/json', body: '{}' })
})
await page.goto('/app/activity')
const card = page.locator('.operation-card').filter({ hasText: 'gemma-3-27b-it' })
await card.locator('.operation-card__pause').click()
await expect.poll(() => requests).toEqual(['/api/operations/job-gemma/pause'])
})
test('separates an unacknowledged failure from the record', async ({ page }) => { test('separates an unacknowledged failure from the record', async ({ page }) => {
await stub(page, { await stub(page, {
operations: [{ operations: [{
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -12,6 +12,8 @@
"timeLeft": "{{value}} left", "timeLeft": "{{value}} left",
"cancel": "Cancel", "cancel": "Cancel",
"cancelLabel": "Cancel {{name}}", "cancelLabel": "Cancel {{name}}",
"pause": "Pause",
"pauseLabel": "Pause {{name}} and keep downloaded data",
"retry": "Retry", "retry": "Retry",
"retryLabel": "Retry {{name}}", "retryLabel": "Retry {{name}}",
"nodeCount": "{{count}} nodes", "nodeCount": "{{count}} nodes",
@@ -29,7 +29,7 @@ function formatEta(seconds) {
return `${Math.floor(minutes / 60)}h ${minutes % 60}m` return `${Math.floor(minutes / 60)}h ${minutes % 60}m`
} }
export default function OperationCard({ operation, onCancel, onDismiss, onRetry }) { export default function OperationCard({ operation, onCancel, onPause, onDismiss, onRetry }) {
const { t } = useTranslation('admin') const { t } = useTranslation('admin')
const nodes = Array.isArray(operation.nodes) ? operation.nodes : [] const nodes = Array.isArray(operation.nodes) ? operation.nodes : []
// Holds only what the user chose. The default has to stay a live // Holds only what the user chose. The default has to stay a live
@@ -144,6 +144,16 @@ export default function OperationCard({ operation, onCancel, onDismiss, onRetry
<div className="operation-card__actions"> <div className="operation-card__actions">
{showProgress && <span className="operation-card__pct" aria-hidden="true">{Math.round(operation.progress)}%</span>} {showProgress && <span className="operation-card__pct" aria-hidden="true">{Math.round(operation.progress)}%</span>}
{canCancel && (
<button
type="button"
className="btn btn-sm btn-secondary operation-card__pause"
onClick={() => onPause?.(operation.jobID)}
aria-label={t('activity.pauseLabel', { name })}
>
{t('activity.pause')}
</button>
)}
{canCancel && ( {canCancel && (
// A page of cards would otherwise hand a screen reader a list of // A page of cards would otherwise hand a screen reader a list of
// identical "Cancel" buttons with nothing to tell them apart. // identical "Cancel" buttons with nothing to tell them apart.
@@ -137,6 +137,16 @@ export function OperationsProvider({ children, pollInterval = 1000 }) {
} }
}, [fetchOperations]) }, [fetchOperations])
const pauseOperation = useCallback(async (jobID) => {
try {
await operationsApi.pause(jobID)
cancelledRef.current.set(jobID, Date.now())
await fetchOperations()
} catch (err) {
setError(err.message)
}
}, [fetchOperations])
// Whether this tab cancelled the job. Read by the strip to tell "the last // Whether this tab cancelled the job. Read by the strip to tell "the last
// operation finished" from "the user called it off": both look identical in // operation finished" from "the user called it off": both look identical in
// /api/operations, which lists neither. // /api/operations, which lists neither.
@@ -226,6 +236,7 @@ export function OperationsProvider({ children, pollInterval = 1000 }) {
fetchHistory, fetchHistory,
clearHistory, clearHistory,
cancelOperation, cancelOperation,
pauseOperation,
wasCancelled, wasCancelled,
dismissFailedOp, dismissFailedOp,
refetch: fetchOperations, refetch: fetchOperations,
+2 -2
View File
@@ -83,7 +83,7 @@ export default function Activity() {
const { t } = useTranslation('admin') const { t } = useTranslation('admin')
const outlet = useOutletContext() const outlet = useOutletContext()
const addToast = outlet?.addToast const addToast = outlet?.addToast
const { operations, history, fetchHistory, clearHistory, cancelOperation, dismissFailedOp } = useOperations() const { operations, history, fetchHistory, clearHistory, cancelOperation, pauseOperation, dismissFailedOp } = useOperations()
const [filter, setFilter] = useState('all') const [filter, setFilter] = useState('all')
useEffect(() => { fetchHistory() }, [fetchHistory]) useEffect(() => { fetchHistory() }, [fetchHistory])
@@ -195,7 +195,7 @@ export default function Activity() {
{t('activity.inProgress')} <span className="activity-section__count">{live.length}</span> {t('activity.inProgress')} <span className="activity-section__count">{live.length}</span>
</h2> </h2>
{live.map((op) => ( {live.map((op) => (
<OperationCard key={op.jobID || op.id} operation={op} onCancel={cancelOperation} /> <OperationCard key={op.jobID || op.id} operation={op} onCancel={cancelOperation} onPause={pauseOperation} />
))} ))}
</section> </section>
)} )}
+1
View File
@@ -173,6 +173,7 @@ export const resourcesApi = {
export const operationsApi = { export const operationsApi = {
list: () => fetchJSON(API_CONFIG.endpoints.operations), list: () => fetchJSON(API_CONFIG.endpoints.operations),
cancel: (jobID) => postJSON(API_CONFIG.endpoints.cancelOperation(jobID), {}), cancel: (jobID) => postJSON(API_CONFIG.endpoints.cancelOperation(jobID), {}),
pause: (jobID) => postJSON(API_CONFIG.endpoints.pauseOperation(jobID), {}),
dismiss: (jobID) => postJSON(API_CONFIG.endpoints.dismissOperation(jobID), {}), dismiss: (jobID) => postJSON(API_CONFIG.endpoints.dismissOperation(jobID), {}),
history: () => fetchJSON(API_CONFIG.endpoints.operationsHistory), history: () => fetchJSON(API_CONFIG.endpoints.operationsHistory),
clearHistory: () => fetchJSON(API_CONFIG.endpoints.operationsHistory, { method: 'DELETE' }), clearHistory: () => fetchJSON(API_CONFIG.endpoints.operationsHistory, { method: 'DELETE' }),
+1
View File
@@ -4,6 +4,7 @@ export const API_CONFIG = {
operations: '/api/operations', operations: '/api/operations',
operationsHistory: '/api/operations/history', operationsHistory: '/api/operations/history',
cancelOperation: (jobID) => `/api/operations/${jobID}/cancel`, cancelOperation: (jobID) => `/api/operations/${jobID}/cancel`,
pauseOperation: (jobID) => `/api/operations/${jobID}/pause`,
dismissOperation: (jobID) => `/api/operations/${jobID}/dismiss`, dismissOperation: (jobID) => `/api/operations/${jobID}/dismiss`,
// Models gallery // Models gallery
+36 -12
View File
@@ -356,6 +356,24 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
}) })
}, adminMiddleware) }, adminMiddleware)
// Pause operation endpoint (admin only). Unlike cancel, pause preserves a
// partial download so submitting the same install later resumes it.
app.POST("/api/operations/:jobID/pause", func(c echo.Context) error {
jobID := c.Param("jobID")
xlog.Debug("API request to pause operation", "jobID", jobID)
if err := galleryService.PauseOperation(jobID); err != nil {
xlog.Error("Failed to pause operation", "error", err, "jobID", jobID)
return c.JSON(http.StatusBadRequest, map[string]any{"error": err.Error()})
}
opcache.DeleteUUID(jobID)
return c.JSON(200, map[string]any{
"success": true,
"message": "Operation paused",
})
}, adminMiddleware)
// Dismiss a failed operation (acknowledge the error and remove it from the list) // Dismiss a failed operation (acknowledge the error and remove it from the list)
app.POST("/api/operations/:jobID/dismiss", func(c echo.Context) error { app.POST("/api/operations/:jobID/dismiss", func(c echo.Context) error {
jobID := c.Param("jobID") jobID := c.Param("jobID")
@@ -970,7 +988,7 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
uid := id.String() uid := id.String()
opcache.Set(galleryID, uid) opcache.Set(galleryID, uid)
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc, pauseFunc := galleryop.NewUserCancellableContext(context.Background())
op := galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{ op := galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{
ID: uid, ID: uid,
GalleryElementName: galleryID, GalleryElementName: galleryID,
@@ -979,9 +997,10 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
BackendGalleries: appConfig.BackendGalleries, BackendGalleries: appConfig.BackendGalleries,
Context: ctx, Context: ctx,
CancelFunc: cancelFunc, CancelFunc: cancelFunc,
PauseFunc: pauseFunc,
} }
// Store cancellation function immediately so queued operations can be cancelled // Store cancellation function immediately so queued operations can be cancelled
galleryService.StoreCancellation(uid, cancelFunc) galleryService.StoreCancellationActions(uid, cancelFunc, pauseFunc)
galleryService.EnqueueModelOp(op) galleryService.EnqueueModelOp(op)
return c.JSON(200, map[string]any{ return c.JSON(200, map[string]any{
@@ -1017,7 +1036,7 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
opcache.Set(galleryID, uid) opcache.Set(galleryID, uid)
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc, pauseFunc := galleryop.NewUserCancellableContext(context.Background())
op := galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{ op := galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{
ID: uid, ID: uid,
Delete: true, Delete: true,
@@ -1026,9 +1045,10 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
BackendGalleries: appConfig.BackendGalleries, BackendGalleries: appConfig.BackendGalleries,
Context: ctx, Context: ctx,
CancelFunc: cancelFunc, CancelFunc: cancelFunc,
PauseFunc: pauseFunc,
} }
// Store cancellation function immediately so queued operations can be cancelled // Store cancellation function immediately so queued operations can be cancelled
galleryService.StoreCancellation(uid, cancelFunc) galleryService.StoreCancellationActions(uid, cancelFunc, pauseFunc)
galleryService.EnqueueModelOp(op) galleryService.EnqueueModelOp(op)
cl.RemoveModelConfig(galleryName) cl.RemoveModelConfig(galleryName)
@@ -1423,19 +1443,20 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
uid := id.String() uid := id.String()
opcache.SetBackend(backendID, uid) opcache.SetBackend(backendID, uid)
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc, pauseFunc := galleryop.NewUserCancellableContext(context.Background())
op := galleryop.ManagementOp[gallery.GalleryBackend, any]{ op := galleryop.ManagementOp[gallery.GalleryBackend, any]{
ID: uid, ID: uid,
GalleryElementName: backendID, GalleryElementName: backendID,
Galleries: appConfig.BackendGalleries, Galleries: appConfig.BackendGalleries,
Context: ctx, Context: ctx,
CancelFunc: cancelFunc, CancelFunc: cancelFunc,
PauseFunc: pauseFunc,
// The React UI's "Reinstall backend" action reuses this route, so // The React UI's "Reinstall backend" action reuses this route, so
// the op must force even when the backend is already installed. // the op must force even when the backend is already installed.
Force: true, Force: true,
} }
// Store cancellation function immediately so queued operations can be cancelled // Store cancellation function immediately so queued operations can be cancelled
galleryService.StoreCancellation(uid, cancelFunc) galleryService.StoreCancellationActions(uid, cancelFunc, pauseFunc)
galleryService.EnqueueBackendOp(op) galleryService.EnqueueBackendOp(op)
return c.JSON(200, map[string]any{ return c.JSON(200, map[string]any{
@@ -1485,19 +1506,20 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
} }
opcache.SetBackend(cacheKey, uid) opcache.SetBackend(cacheKey, uid)
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc, pauseFunc := galleryop.NewUserCancellableContext(context.Background())
op := galleryop.ManagementOp[gallery.GalleryBackend, any]{ op := galleryop.ManagementOp[gallery.GalleryBackend, any]{
ID: uid, ID: uid,
GalleryElementName: req.Name, // May be empty, will be derived during installation GalleryElementName: req.Name, // May be empty, will be derived during installation
Galleries: appConfig.BackendGalleries, Galleries: appConfig.BackendGalleries,
Context: ctx, Context: ctx,
CancelFunc: cancelFunc, CancelFunc: cancelFunc,
PauseFunc: pauseFunc,
ExternalURI: req.URI, ExternalURI: req.URI,
ExternalName: req.Name, ExternalName: req.Name,
ExternalAlias: req.Alias, ExternalAlias: req.Alias,
} }
// Store cancellation function immediately so queued operations can be cancelled // Store cancellation function immediately so queued operations can be cancelled
galleryService.StoreCancellation(uid, cancelFunc) galleryService.StoreCancellationActions(uid, cancelFunc, pauseFunc)
galleryService.EnqueueBackendOp(op) galleryService.EnqueueBackendOp(op)
return c.JSON(200, map[string]any{ return c.JSON(200, map[string]any{
@@ -1533,7 +1555,7 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
opcache.SetBackend(backendID, uid) opcache.SetBackend(backendID, uid)
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc, pauseFunc := galleryop.NewUserCancellableContext(context.Background())
op := galleryop.ManagementOp[gallery.GalleryBackend, any]{ op := galleryop.ManagementOp[gallery.GalleryBackend, any]{
ID: uid, ID: uid,
Delete: true, Delete: true,
@@ -1541,9 +1563,10 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
Galleries: appConfig.BackendGalleries, Galleries: appConfig.BackendGalleries,
Context: ctx, Context: ctx,
CancelFunc: cancelFunc, CancelFunc: cancelFunc,
PauseFunc: pauseFunc,
} }
// Store cancellation function immediately so queued operations can be cancelled // Store cancellation function immediately so queued operations can be cancelled
galleryService.StoreCancellation(uid, cancelFunc) galleryService.StoreCancellationActions(uid, cancelFunc, pauseFunc)
galleryService.EnqueueBackendOp(op) galleryService.EnqueueBackendOp(op)
return c.JSON(200, map[string]any{ return c.JSON(200, map[string]any{
@@ -1652,7 +1675,7 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
// and the Backends UI can reflect progress on the affected row. // and the Backends UI can reflect progress on the affected row.
opcache.SetBackend(backendName, uid) opcache.SetBackend(backendName, uid)
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc, pauseFunc := galleryop.NewUserCancellableContext(context.Background())
op := galleryop.ManagementOp[gallery.GalleryBackend, any]{ op := galleryop.ManagementOp[gallery.GalleryBackend, any]{
ID: uid, ID: uid,
GalleryElementName: backendName, GalleryElementName: backendName,
@@ -1660,9 +1683,10 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
Upgrade: true, Upgrade: true,
Context: ctx, Context: ctx,
CancelFunc: cancelFunc, CancelFunc: cancelFunc,
PauseFunc: pauseFunc,
} }
// Store cancellation function immediately so queued operations can be cancelled // Store cancellation function immediately so queued operations can be cancelled
galleryService.StoreCancellation(uid, cancelFunc) galleryService.StoreCancellationActions(uid, cancelFunc, pauseFunc)
galleryService.EnqueueBackendOp(op) galleryService.EnqueueBackendOp(op)
return c.JSON(200, map[string]any{ return c.JSON(200, map[string]any{
@@ -339,6 +339,33 @@ var _ = Describe("/api/operations with node-scoped backend ops", func() {
Expect(envelope.Operations[0]).ToNot(HaveKey("isCancelled")) Expect(envelope.Operations[0]).ToNot(HaveKey("isCancelled"))
}) })
It("pauses through the resume-safe operation callback", func() {
state, err := system.GetSystemState(system.WithModelPath(GinkgoT().TempDir()))
Expect(err).NotTo(HaveOccurred())
appCfg := &config.ApplicationConfig{SystemState: state}
galleryService := galleryop.NewGalleryService(appCfg, nil)
opcache := galleryop.NewOpCache(galleryService)
opcache.Set("localai@gemma", "job-pause")
var cancelled, paused bool
galleryService.StoreCancellationActions(
"job-pause",
func() { cancelled = true },
func() { paused = true },
)
e := echo.New()
routes.RegisterUIAPIRoutes(e, nil, nil, appCfg, galleryService, opcache, &application.Application{}, noopMw)
req := httptest.NewRequest(http.MethodPost, "/api/operations/job-pause/pause", nil)
rec := httptest.NewRecorder()
e.ServeHTTP(rec, req)
Expect(rec.Code).To(Equal(http.StatusOK))
Expect(paused).To(BeTrue())
Expect(cancelled).To(BeFalse())
Expect(opcache.Get("localai@gemma")).To(BeEmpty())
})
It("reports a running removal as a deletion", func() { It("reports a running removal as a deletion", func() {
state, err := system.GetSystemState(system.WithModelPath(GinkgoT().TempDir())) state, err := system.GetSystemState(system.WithModelPath(GinkgoT().TempDir()))
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
@@ -54,6 +54,22 @@ var _ = Describe("GalleryService.CancelOperation persistence", func() {
Expect(fresh.GetStatus("op-cancel")).To(BeNil(), Expect(fresh.GetStatus("op-cancel")).To(BeNil(),
"a cancelled op must not hydrate back as active after a restart") "a cancelled op must not hydrate back as active after a restart")
}) })
It("pauses with the resume-safe callback instead of the destructive cancel callback", func() {
svc := galleryop.NewGalleryService(&config.ApplicationConfig{}, nil)
var cancelled, paused bool
svc.StoreCancellationActions("op-pause", func() { cancelled = true }, func() { paused = true })
Expect(svc.PauseOperation("op-pause")).To(Succeed())
Expect(paused).To(BeTrue())
Expect(cancelled).To(BeFalse())
status := svc.GetStatus("op-pause")
Expect(status).ToNot(BeNil())
Expect(status.Processed).To(BeTrue())
Expect(status.Cancelled).To(BeTrue())
Expect(status.Message).To(Equal("paused"))
})
}) })
// Reproduces "an op orphaned by a replica that died mid-flight stays 'pending' // Reproduces "an op orphaned by a replica that died mid-flight stays 'pending'
+2
View File
@@ -32,6 +32,7 @@ type ManagementOp[T any, E any] struct {
// Context for cancellation support // Context for cancellation support
Context context.Context Context context.Context
CancelFunc context.CancelFunc CancelFunc context.CancelFunc
PauseFunc context.CancelFunc
// External backend installation parameters (for OCI/URL/path) // External backend installation parameters (for OCI/URL/path)
// These are used when installing backends from external sources rather than galleries // These are used when installing backends from external sources rather than galleries
@@ -189,6 +190,7 @@ type GalleryProgressEvent struct {
// runs the cancel func on whichever replica registered it. // runs the cancel func on whichever replica registered it.
type GalleryCancelEvent struct { type GalleryCancelEvent struct {
JobID string `json:"id"` JobID string `json:"id"`
Pause bool `json:"pause,omitempty"`
} }
// NodeStatus values shared between NodeProgress (per-node tick) and the // NodeStatus values shared between NodeProgress (per-node tick) and the
+63 -27
View File
@@ -28,7 +28,7 @@ type GalleryService struct {
modelManager ModelManager modelManager ModelManager
backendManager BackendManager backendManager BackendManager
statuses map[string]*OpStatus statuses map[string]*OpStatus
cancellations map[string]context.CancelFunc cancellations map[string]cancellationActions
// Distributed mode (nil when not in distributed mode). // Distributed mode (nil when not in distributed mode).
// natsClient is the wider MessagingClient (Publisher + subscribe methods) // natsClient is the wider MessagingClient (Publisher + subscribe methods)
@@ -67,7 +67,7 @@ func NewGalleryService(appConfig *config.ApplicationConfig, ml *model.ModelLoade
modelManager: NewLocalModelManager(appConfig, ml), modelManager: NewLocalModelManager(appConfig, ml),
backendManager: NewLocalBackendManager(appConfig, ml), backendManager: NewLocalBackendManager(appConfig, ml),
statuses: make(map[string]*OpStatus), statuses: make(map[string]*OpStatus),
cancellations: make(map[string]context.CancelFunc), cancellations: make(map[string]cancellationActions),
} }
} }
@@ -387,6 +387,17 @@ func (g *GalleryService) failStaleStatus(id string) {
// SubjectGalleryCancelWildcard subscriber and runs it locally. The caller // SubjectGalleryCancelWildcard subscriber and runs it locally. The caller
// gets a non-error reply so the UI shows the cancel as accepted. // gets a non-error reply so the UI shows the cancel as accepted.
func (g *GalleryService) CancelOperation(id string) error { func (g *GalleryService) CancelOperation(id string) error {
return g.stopOperation(id, false)
}
// PauseOperation stops an in-progress download while preserving its partial
// file. Re-submitting the same install resumes it through the downloader's
// existing HTTP Range support.
func (g *GalleryService) PauseOperation(id string) error {
return g.stopOperation(id, true)
}
func (g *GalleryService) stopOperation(id string, pause bool) error {
g.Lock() g.Lock()
if status, ok := g.statuses[id]; ok && status.Cancelled { if status, ok := g.statuses[id]; ok && status.Cancelled {
@@ -394,7 +405,7 @@ func (g *GalleryService) CancelOperation(id string) error {
return fmt.Errorf("operation %q is already cancelled", id) return fmt.Errorf("operation %q is already cancelled", id)
} }
cancelFunc, localExists := g.cancellations[id] actions, localExists := g.cancellations[id]
if localExists { if localExists {
delete(g.cancellations, id) delete(g.cancellations, id)
} }
@@ -410,12 +421,12 @@ func (g *GalleryService) CancelOperation(id string) error {
if status, ok := g.statuses[id]; ok { if status, ok := g.statuses[id]; ok {
status.Cancelled = true status.Cancelled = true
status.Processed = true status.Processed = true
status.Message = "cancelled" status.Message = map[bool]string{true: "paused", false: "cancelled"}[pause]
} else { } else {
g.statuses[id] = &OpStatus{ g.statuses[id] = &OpStatus{
Cancelled: true, Cancelled: true,
Processed: true, Processed: true,
Message: "cancelled", Message: map[bool]string{true: "paused", false: "cancelled"}[pause],
Cancellable: false, Cancellable: false,
} }
} }
@@ -435,11 +446,15 @@ func (g *GalleryService) CancelOperation(id string) error {
// I/O and user-provided callback after Unlock — the cancel-wildcard // I/O and user-provided callback after Unlock — the cancel-wildcard
// subscriber loops back into applyCancel on this same replica, which // subscriber loops back into applyCancel on this same replica, which
// would otherwise deadlock on g.Mutex. // would otherwise deadlock on g.Mutex.
if cancelFunc != nil { stopFunc := actions.cancel
cancelFunc() if pause {
stopFunc = actions.pause
}
if stopFunc != nil {
stopFunc()
} }
if nc != nil { if nc != nil {
if err := nc.Publish(messaging.SubjectGalleryCancel(id), GalleryCancelEvent{JobID: id}); err != nil { if err := nc.Publish(messaging.SubjectGalleryCancel(id), GalleryCancelEvent{JobID: id, Pause: pause}); err != nil {
xlog.Warn("Failed to broadcast gallery cancel", "op_id", id, "error", err) xlog.Warn("Failed to broadcast gallery cancel", "op_id", id, "error", err)
} }
} }
@@ -452,9 +467,9 @@ func (g *GalleryService) CancelOperation(id string) error {
// run the local cancel func if we have one (no echo via NATS), and reflect // run the local cancel func if we have one (no echo via NATS), and reflect
// the cancellation in the local statuses map. Idempotent: a replica that // the cancellation in the local statuses map. Idempotent: a replica that
// already cancelled this op locally treats the inbound event as a no-op. // already cancelled this op locally treats the inbound event as a no-op.
func (g *GalleryService) applyCancel(id string) { func (g *GalleryService) applyCancel(id string, pause bool) {
g.Lock() g.Lock()
cancelFunc, hasCancel := g.cancellations[id] actions, hasCancel := g.cancellations[id]
if hasCancel { if hasCancel {
delete(g.cancellations, id) delete(g.cancellations, id)
} }
@@ -465,12 +480,12 @@ func (g *GalleryService) applyCancel(id string) {
} }
status.Cancelled = true status.Cancelled = true
status.Processed = true status.Processed = true
status.Message = "cancelled" status.Message = map[bool]string{true: "paused", false: "cancelled"}[pause]
} else { } else {
g.statuses[id] = &OpStatus{ g.statuses[id] = &OpStatus{
Cancelled: true, Cancelled: true,
Processed: true, Processed: true,
Message: "cancelled", Message: map[bool]string{true: "paused", false: "cancelled"}[pause],
Cancellable: false, Cancellable: false,
} }
} }
@@ -478,8 +493,12 @@ func (g *GalleryService) applyCancel(id string) {
// Invoke the cancel func after Unlock so a callback that touches // Invoke the cancel func after Unlock so a callback that touches
// GalleryService doesn't re-enter the mutex. // GalleryService doesn't re-enter the mutex.
if hasCancel { stopFunc := actions.cancel
cancelFunc() if pause {
stopFunc = actions.pause
}
if hasCancel && stopFunc != nil {
stopFunc()
} }
} }
@@ -488,23 +507,40 @@ func (g *GalleryService) applyCancel(id string) {
// distinguish a deliberate user cancel (discard the half-downloaded .partial) // distinguish a deliberate user cancel (discard the half-downloaded .partial)
// from an incidental cancellation such as process shutdown (keep the .partial // from an incidental cancellation such as process shutdown (keep the .partial
// so the next run resumes via Range instead of restarting from zero). // so the next run resumes via Range instead of restarting from zero).
func newUserCancellableContext(parent context.Context) (context.Context, context.CancelFunc) { // NewUserCancellableContext creates distinct callbacks for destructive cancel
// and resume-safe pause while sharing one operation context.
func NewUserCancellableContext(parent context.Context) (context.Context, context.CancelFunc, context.CancelFunc) {
ctx, cancelCause := context.WithCancelCause(parent) ctx, cancelCause := context.WithCancelCause(parent)
return ctx, func() { cancelCause(downloader.ErrUserCancelled) } return ctx,
func() { cancelCause(downloader.ErrUserCancelled) },
func() { cancelCause(context.Canceled) }
} }
// storeCancellation stores a cancellation function for an operation type cancellationActions struct {
func (g *GalleryService) storeCancellation(id string, cancelFunc context.CancelFunc) { cancel context.CancelFunc
pause context.CancelFunc
}
func (g *GalleryService) storeCancellation(id string, cancelFunc, pauseFunc context.CancelFunc) {
g.Lock() g.Lock()
defer g.Unlock() defer g.Unlock()
g.cancellations[id] = cancelFunc if pauseFunc == nil {
pauseFunc = cancelFunc
}
g.cancellations[id] = cancellationActions{cancel: cancelFunc, pause: pauseFunc}
} }
// StoreCancellation is a public method to store a cancellation function for an operation // StoreCancellation is a public method to store a cancellation function for an operation
// This allows cancellation functions to be stored immediately when operations are created, // This allows cancellation functions to be stored immediately when operations are created,
// enabling cancellation of queued operations that haven't started processing yet. // enabling cancellation of queued operations that haven't started processing yet.
func (g *GalleryService) StoreCancellation(id string, cancelFunc context.CancelFunc) { func (g *GalleryService) StoreCancellation(id string, cancelFunc context.CancelFunc) {
g.storeCancellation(id, cancelFunc) g.storeCancellation(id, cancelFunc, cancelFunc)
}
// StoreCancellationActions registers distinct destructive-cancel and
// resume-safe pause callbacks for an operation.
func (g *GalleryService) StoreCancellationActions(id string, cancelFunc, pauseFunc context.CancelFunc) {
g.storeCancellation(id, cancelFunc, pauseFunc)
} }
// removeCancellation removes a cancellation function when operation completes // removeCancellation removes a cancellation function when operation completes
@@ -554,10 +590,10 @@ func (g *GalleryService) Start(c context.Context, cl *config.ModelConfigLoader,
case op := <-g.BackendGalleryChannel: case op := <-g.BackendGalleryChannel:
// Create context if not provided // Create context if not provided
if op.Context == nil { if op.Context == nil {
op.Context, op.CancelFunc = newUserCancellableContext(c) op.Context, op.CancelFunc, op.PauseFunc = NewUserCancellableContext(c)
g.storeCancellation(op.ID, op.CancelFunc) g.storeCancellation(op.ID, op.CancelFunc, op.PauseFunc)
} else if op.CancelFunc != nil { } else if op.CancelFunc != nil {
g.storeCancellation(op.ID, op.CancelFunc) g.storeCancellation(op.ID, op.CancelFunc, op.PauseFunc)
} }
// Create DB record for distributed tracking // Create DB record for distributed tracking
if g.galleryStore != nil { if g.galleryStore != nil {
@@ -597,10 +633,10 @@ func (g *GalleryService) Start(c context.Context, cl *config.ModelConfigLoader,
case op := <-g.ModelGalleryChannel: case op := <-g.ModelGalleryChannel:
// Create context if not provided // Create context if not provided
if op.Context == nil { if op.Context == nil {
op.Context, op.CancelFunc = newUserCancellableContext(c) op.Context, op.CancelFunc, op.PauseFunc = NewUserCancellableContext(c)
g.storeCancellation(op.ID, op.CancelFunc) g.storeCancellation(op.ID, op.CancelFunc, op.PauseFunc)
} else if op.CancelFunc != nil { } else if op.CancelFunc != nil {
g.storeCancellation(op.ID, op.CancelFunc) g.storeCancellation(op.ID, op.CancelFunc, op.PauseFunc)
} }
// Create DB record for distributed tracking // Create DB record for distributed tracking
if g.galleryStore != nil { if g.galleryStore != nil {
@@ -663,7 +699,7 @@ func (g *GalleryService) SubscribeBroadcasts() error {
if evt.JobID == "" { if evt.JobID == "" {
return return
} }
g.applyCancel(evt.JobID) g.applyCancel(evt.JobID, evt.Pause)
}) })
if err != nil { if err != nil {
if uerr := progressSub.Unsubscribe(); uerr != nil { if uerr := progressSub.Unsubscribe(); uerr != nil {
+1 -1
View File
@@ -157,7 +157,7 @@ When authentication is enabled, the following endpoints require admin role:
**Model & Backend Management:** **Model & Backend Management:**
- `GET /api/models`, `POST /api/models/install/*`, `POST /api/models/delete/*` - `GET /api/models`, `POST /api/models/install/*`, `POST /api/models/delete/*`
- `GET /api/backends`, `POST /api/backends/install/*`, `POST /api/backends/delete/*` - `GET /api/backends`, `POST /api/backends/install/*`, `POST /api/backends/delete/*`
- `GET /api/operations`, `POST /api/operations/*/cancel`, `POST /api/operations/*/dismiss` - `GET /api/operations`, `POST /api/operations/*/cancel`, `POST /api/operations/*/pause`, `POST /api/operations/*/dismiss`
- `GET /api/operations/history`, `DELETE /api/operations/history` - `GET /api/operations/history`, `DELETE /api/operations/history`
- `GET /models/available`, `GET /models/galleries`, `GET /models/jobs/*` - `GET /models/available`, `GET /models/galleries`, `GET /models/jobs/*`
- `GET /backends`, `GET /backends/available`, `GET /backends/galleries` - `GET /backends`, `GET /backends/available`, `GET /backends/galleries`
+5
View File
@@ -170,10 +170,15 @@ Both surfaces are backed by these endpoints. All of them are admin-only when
| -------- | -------------------------------- | ------------------------------------------------------------------------ | | -------- | -------------------------------- | ------------------------------------------------------------------------ |
| `GET` | `/api/operations` | Running, queued and failed operations, least advanced first. | | `GET` | `/api/operations` | Running, queued and failed operations, least advanced first. |
| `POST` | `/api/operations/{jobID}/cancel` | Cancel a queued operation, or a running install. Not a running removal. | | `POST` | `/api/operations/{jobID}/cancel` | Cancel a queued operation, or a running install. Not a running removal. |
| `POST` | `/api/operations/{jobID}/pause` | Pause a queued operation or running download and preserve partial data. |
| `POST` | `/api/operations/{jobID}/dismiss`| Acknowledge a failed operation and move it into the record. | | `POST` | `/api/operations/{jobID}/dismiss`| Acknowledge a failed operation and move it into the record. |
| `GET` | `/api/operations/history` | List finished operations, newest first. | | `GET` | `/api/operations/history` | List finished operations, newest first. |
| `DELETE` | `/api/operations/history` | Clear the record. Live operations are untouched. | | `DELETE` | `/api/operations/history` | Clear the record. Live operations are untouched. |
Pausing preserves the download's `.partial` file. Start the same model or
backend installation again to resume from the saved bytes when the origin
supports HTTP Range requests. Cancelling intentionally discards partial data.
Both `GET` endpoints wrap their list in an `operations` key rather than Both `GET` endpoints wrap their list in an `operations` key rather than
returning a bare array: returning a bare array: