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
74 lines
2.0 KiB
Go
74 lines
2.0 KiB
Go
package work
|
||
|
||
import (
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"sync"
|
||
"testing"
|
||
|
||
jsoniter "github.com/json-iterator/go"
|
||
"github.com/stretchr/testify/require"
|
||
)
|
||
|
||
// TestAtomicWriteFileStress applies extreme concurrent pressure to verify robustness
|
||
func TestAtomicWriteFileStress(t *testing.T) {
|
||
if testing.Short() {
|
||
t.Skip("Skipping stress test in short mode")
|
||
}
|
||
|
||
tmpDir := t.TempDir()
|
||
targetFile := filepath.Join(tmpDir, "stress.json")
|
||
|
||
// Large data to increase write time and chances of race conditions
|
||
largeData := make([]byte, 1024*10) // 10KB
|
||
for i := range largeData {
|
||
largeData[i] = byte('A' + (i % 26))
|
||
}
|
||
testData := []byte(fmt.Sprintf(`{"large_data": "%s", "id": "test"}`, string(largeData)))
|
||
|
||
// High concurrency stress test
|
||
var wg sync.WaitGroup
|
||
numGoroutines := 100
|
||
writesPerGoroutine := 5
|
||
|
||
for i := range numGoroutines {
|
||
wg.Add(1)
|
||
go func(goroutineID int) {
|
||
defer wg.Done()
|
||
|
||
for j := range writesPerGoroutine {
|
||
// Use different temp prefixes to test CreateTemp uniqueness
|
||
prefix := fmt.Sprintf("stress_%d_%d.json", goroutineID, j)
|
||
err := atomicWriteFile(testData, targetFile, prefix)
|
||
require.NoError(t, err)
|
||
|
||
// No sleep - maximum pressure
|
||
}
|
||
}(i)
|
||
}
|
||
|
||
wg.Wait()
|
||
|
||
// Verify final file integrity
|
||
finalData, err := os.ReadFile(targetFile)
|
||
require.NoError(t, err)
|
||
require.NotEmpty(t, finalData, "Final file should not be empty")
|
||
|
||
// Should be valid JSON
|
||
var result map[string]any
|
||
err = jsoniter.Unmarshal(finalData, &result)
|
||
require.NoError(t, err, "Final file should contain valid JSON after stress test")
|
||
|
||
// Verify it's our expected data
|
||
require.Equal(t, "test", result["id"], "Final file should contain our test data")
|
||
|
||
// Critical: No temp files should remain after stress test
|
||
files, err := filepath.Glob(filepath.Join(tmpDir, "*tmp*"))
|
||
require.NoError(t, err)
|
||
require.Empty(t, files, "No temporary files should remain after stress test - all should be cleaned up")
|
||
|
||
t.Logf("Stress test completed: %d goroutines × %d writes = %d total operations",
|
||
numGoroutines, writesPerGoroutine, numGoroutines*writesPerGoroutine)
|
||
}
|