diff --git a/weed/admin/plugin/job_queue.go b/weed/admin/plugin/job_queue.go
new file mode 100644
index 000000000..823646cb7
--- /dev/null
+++ b/weed/admin/plugin/job_queue.go
@@ -0,0 +1,298 @@
+package plugin
+
+import (
+	"container/heap"
+	"fmt"
+	"sync"
+	"time"
+)
+
+// JobQueue manages job queueing with priority, deduplication, retry and history
+type JobQueue struct {
+	mu              sync.RWMutex
+	priorityQueue   *PriorityQueue
+	seenJobs        map[string]bool // For deduplication
+	jobHistory      []*ExecutionRecord
+	maxHistorySize  int
+	deduplicationTTL time.Duration
+	lastSeenJob     map[string]time.Time
+}
+
+// PriorityQueue implements heap.Interface for job ordering
+type PriorityQueue []*Job
+
+func (pq PriorityQueue) Len() int { return len(pq) }
+
+func (pq PriorityQueue) Less(i, j int) bool {
+	// Higher priority jobs come first
+	if pq[i].Type != pq[j].Type {
+		return pq[i].Type < pq[j].Type
+	}
+	// If same type, earlier creation time comes first
+	return pq[i].CreatedAt.Before(pq[j].CreatedAt)
+}
+
+func (pq PriorityQueue) Swap(i, j int) {
+	pq[i], pq[j] = pq[j], pq[i]
+}
+
+func (pq *PriorityQueue) Push(x interface{}) {
+	*pq = append(*pq, x.(*Job))
+}
+
+func (pq *PriorityQueue) Pop() interface{} {
+	old := *pq
+	n := len(old)
+	item := old[n-1]
+	*pq = old[0 : n-1]
+	return item
+}
+
+// NewJobQueue creates a new job queue
+func NewJobQueue(maxHistorySize int, deduplicationTTL time.Duration) *JobQueue {
+	jq := &JobQueue{
+		priorityQueue:    &PriorityQueue{},
+		seenJobs:         make(map[string]bool),
+		jobHistory:       make([]*ExecutionRecord, 0, maxHistorySize),
+		maxHistorySize:   maxHistorySize,
+		deduplicationTTL: deduplicationTTL,
+		lastSeenJob:      make(map[string]time.Time),
+	}
+	heap.Init(jq.priorityQueue)
+	return jq
+}
+
+// Enqueue adds a job to the queue with deduplication
+func (jq *JobQueue) Enqueue(job *Job) error {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	// Check for duplicates within TTL window
+	if lastSeen, exists := jq.lastSeenJob[job.ID]; exists {
+		if time.Since(lastSeen) < jq.deduplicationTTL {
+			return fmt.Errorf("job %s already enqueued recently", job.ID)
+		}
+	}
+
+	job.SetState(JobStatePending)
+	heap.Push(jq.priorityQueue, job)
+	jq.seenJobs[job.ID] = true
+	jq.lastSeenJob[job.ID] = time.Now()
+
+	return nil
+}
+
+// Dequeue retrieves the next job from the queue
+func (jq *JobQueue) Dequeue() *Job {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	if jq.priorityQueue.Len() == 0 {
+		return nil
+	}
+
+	job := heap.Pop(jq.priorityQueue).(*Job)
+	return job
+}
+
+// Peek returns the next job without removing it
+func (jq *JobQueue) Peek() *Job {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+
+	if jq.priorityQueue.Len() == 0 {
+		return nil
+	}
+
+	return (*jq.priorityQueue)[0]
+}
+
+// Size returns the current queue size
+func (jq *JobQueue) Size() int {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+	return jq.priorityQueue.Len()
+}
+
+// RecordExecution adds an execution record to history
+func (jq *JobQueue) RecordExecution(record *ExecutionRecord) {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	jq.jobHistory = append(jq.jobHistory, record)
+
+	// Keep history size bounded
+	if len(jq.jobHistory) > jq.maxHistorySize {
+		// Remove oldest entries
+		removeCount := len(jq.jobHistory) - jq.maxHistorySize
+		jq.jobHistory = jq.jobHistory[removeCount:]
+	}
+}
+
+// GetHistory returns job execution history
+func (jq *JobQueue) GetHistory(limit int) []*ExecutionRecord {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+
+	if limit <= 0 || limit > len(jq.jobHistory) {
+		limit = len(jq.jobHistory)
+	}
+
+	// Return the most recent entries
+	startIdx := len(jq.jobHistory) - limit
+	if startIdx < 0 {
+		startIdx = 0
+	}
+
+	result := make([]*ExecutionRecord, limit)
+	copy(result, jq.jobHistory[startIdx:])
+	return result
+}
+
+// GetHistoryForPlugin returns history for a specific plugin
+func (jq *JobQueue) GetHistoryForPlugin(pluginID string, limit int) []*ExecutionRecord {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+
+	var result []*ExecutionRecord
+	for i := len(jq.jobHistory) - 1; i >= 0 && len(result) < limit; i-- {
+		if jq.jobHistory[i].PluginID == pluginID {
+			result = append(result, jq.jobHistory[i])
+		}
+	}
+	return result
+}
+
+// GetHistoryForJobType returns history for a specific job type
+func (jq *JobQueue) GetHistoryForJobType(jobType string, limit int) []*ExecutionRecord {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+
+	var result []*ExecutionRecord
+	for i := len(jq.jobHistory) - 1; i >= 0 && len(result) < limit; i-- {
+		if jq.jobHistory[i].JobType == jobType {
+			result = append(result, jq.jobHistory[i])
+		}
+	}
+	return result
+}
+
+// ClearHistory removes all execution history
+func (jq *JobQueue) ClearHistory() {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+	jq.jobHistory = make([]*ExecutionRecord, 0, jq.maxHistorySize)
+}
+
+// PurgeOldHistory removes history entries older than the specified time
+func (jq *JobQueue) PurgeOldHistory(beforeTime time.Time) int {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	removed := 0
+	newHistory := make([]*ExecutionRecord, 0)
+
+	for _, record := range jq.jobHistory {
+		if record.CreatedAt.After(beforeTime) {
+			newHistory = append(newHistory, record)
+		} else {
+			removed++
+		}
+	}
+
+	jq.jobHistory = newHistory
+	return removed
+}
+
+// HistorySize returns the number of records in history
+func (jq *JobQueue) HistorySize() int {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+	return len(jq.jobHistory)
+}
+
+// RetryJob re-enqueues a failed job up to maxRetries times
+func (jq *JobQueue) RetryJob(job *Job, maxRetries int) error {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	if job.RetryCount >= maxRetries {
+		return fmt.Errorf("job %s exceeded max retries (%d)", job.ID, maxRetries)
+	}
+
+	job.RetryCount++
+	job.SetState(JobStatePending)
+	heap.Push(jq.priorityQueue, job)
+
+	return nil
+}
+
+// GetExecutionStats returns statistics about job executions
+func (jq *JobQueue) GetExecutionStats() map[string]interface{} {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+
+	completed := 0
+	failed := 0
+	totalExecutionTime := int64(0)
+
+	for _, record := range jq.jobHistory {
+		switch record.State {
+		case JobStateCompleted:
+			completed++
+		case JobStateFailed:
+			failed++
+		}
+		if record.CompletedAt != nil && record.StartedAt != nil {
+			totalExecutionTime += record.CompletedAt.Sub(*record.StartedAt).Milliseconds()
+		}
+	}
+
+	avgExecutionTime := int64(0)
+	if completed+failed > 0 {
+		avgExecutionTime = totalExecutionTime / int64(completed+failed)
+	}
+
+	return map[string]interface{}{
+		"total_history":          len(jq.jobHistory),
+		"completed_jobs":         completed,
+		"failed_jobs":            failed,
+		"avg_execution_time_ms":  avgExecutionTime,
+		"current_queue_size":     jq.priorityQueue.Len(),
+	}
+}
+
+// GetQueuedJobs returns all jobs currently in the queue
+func (jq *JobQueue) GetQueuedJobs() []*Job {
+	jq.mu.RLock()
+	defer jq.mu.RUnlock()
+
+	result := make([]*Job, len(*jq.priorityQueue))
+	copy(result, *jq.priorityQueue)
+	return result
+}
+
+// RemoveJob removes a specific job from the queue
+func (jq *JobQueue) RemoveJob(jobID string) bool {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	for i, job := range *jq.priorityQueue {
+		if job.ID == jobID {
+			heap.Remove(jq.priorityQueue, i)
+			return true
+		}
+	}
+	return false
+}
+
+// PurgeQueuedJobs clears all pending jobs from the queue
+func (jq *JobQueue) PurgeQueuedJobs() int {
+	jq.mu.Lock()
+	defer jq.mu.Unlock()
+
+	count := jq.priorityQueue.Len()
+	*jq.priorityQueue = PriorityQueue{}
+	heap.Init(jq.priorityQueue)
+	return count
+}