mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Zach LeslieandGitHub
c5fe2e7e6b
* [backend-scheduler] initial work file sharding for more targeted marshaling * Add integration coverage for both work implementations * Update config manifest * Add integration coverage for sharded work configuration * Fix integration test config path * Add test coverage for sharded work migrations and marshaling * Use new sharded work file in the backend and handle migration * Clean up backend after migration * Use shared method for shard file name * Add method to write files atomically and test * Update work interface for simpler flush/load interactions * Update callers for local flush/load interface methods * Fix lint * Fix lint * Fix lint * Clean up * Clean up after refactor new interface functions * Promote the sharded implementation to be the only implementation * Clean up comments * Generate config manifest * Clean up interface and improve test coverage * Drop transitional comments * Protect work marshal/unmarshal with a mutex * Add test coverage for concurrent marshal/unmarshal * Lint * Fix race condition in Marshal/Unmarshal when jobs are operating * Relocate methods * Update changelog * Drop interface method only used in tests * Rename method for clarity, add doc * Update doc for clarity * Lock for safety * Clarify and improve locking flush behavior * Encapsulate the lock/unlock in a func * Address some PR feedback * Further unifi shard file name generation * Modify work to use int rather than uint8 * Fix lint
49 lines
1.1 KiB
Go
49 lines
1.1 KiB
Go
package work
|
|
|
|
import (
|
|
"context"
|
|
)
|
|
|
|
// Interface defines the common interface for work management
|
|
type Interface interface {
|
|
// Job management
|
|
AddJob(j *Job) error
|
|
StartJob(id string)
|
|
GetJob(id string) *Job
|
|
RemoveJob(id string)
|
|
CompleteJob(id string)
|
|
FailJob(id string)
|
|
SetJobCompactionOutput(id string, output []string)
|
|
|
|
// Job queries
|
|
ListJobs() []*Job
|
|
GetJobForWorker(ctx context.Context, workerID string) *Job
|
|
|
|
// Maintenance
|
|
Prune(ctx context.Context)
|
|
|
|
// Serialization
|
|
Marshal() ([]byte, error)
|
|
Unmarshal(data []byte) error
|
|
|
|
// Local file operations
|
|
FlushToLocal(ctx context.Context, localPath string, affectedJobIDs []string) error
|
|
LoadFromLocal(ctx context.Context, localPath string) error
|
|
}
|
|
|
|
// ShardedWorkInterface extends WorkInterface with sharding-specific methods
|
|
type ShardedWorkInterface interface {
|
|
Interface
|
|
|
|
// Sharding-specific optimizations
|
|
MarshalShard(shardID int) ([]byte, error)
|
|
UnmarshalShard(shardID int, data []byte) error
|
|
GetShardStats() map[string]any
|
|
GetShardID(jobID string) int
|
|
}
|
|
|
|
var (
|
|
_ Interface = (*Work)(nil)
|
|
_ ShardedWorkInterface = (*Work)(nil)
|
|
)
|