Compare commits
3 Commits
add-readme
...
84f8366c41
| Author | SHA1 | Date | |
|---|---|---|---|
| 84f8366c41 | |||
| 2645f60536 | |||
| a256b83734 |
41
TODO.md
Normal file
41
TODO.md
Normal file
@@ -0,0 +1,41 @@
|
|||||||
|
# Workflow
|
||||||
|
|
||||||
|
* branch (from `main`)
|
||||||
|
* do the work in Next Step
|
||||||
|
* move Next Step to the top of Completed Steps
|
||||||
|
* move the top item of Future Steps into Next Step
|
||||||
|
* commit (`TODO.md` changes in the same commit as the work)
|
||||||
|
* merge to `main` if the branch is not protected, otherwise open a PR
|
||||||
|
* push
|
||||||
|
|
||||||
|
# Status
|
||||||
|
|
||||||
|
pre-1.0
|
||||||
|
|
||||||
|
# Next Step
|
||||||
|
|
||||||
|
Bring the repo into policy compliance in one commit: add .gitignore,
|
||||||
|
.dockerignore, .editorconfig, and .golangci.yml. Verify `make check` stays
|
||||||
|
green with the new lint config.
|
||||||
|
|
||||||
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-06-28: Fixed errcheck lint failures; added compilation smoke test;
|
||||||
|
tidied go.mod.
|
||||||
|
- 2026-06-28: Added repo scaffolding: README, LICENSE, Makefile,
|
||||||
|
Dockerfile, REPO_POLICIES.md, and Gitea CI.
|
||||||
|
- 2026-02-12: Fixed SQLite database locking by removing parallel
|
||||||
|
processing; fixed Linux build via golang.org/x/sys/unix Fadvise.
|
||||||
|
- 2026-02-12: Optimized file copy for large databases; moved temp
|
||||||
|
directory to NVMe scratch storage.
|
||||||
|
- 2026-02-11: Added date range support.
|
||||||
|
- 2026-02-09: Initial implementation: single-day extraction, specific-date
|
||||||
|
targeting, faster pruning of throwaway database copies.
|
||||||
|
|
||||||
|
# Future Steps
|
||||||
|
|
||||||
|
- Add .gitignore, .dockerignore, .editorconfig, .golangci.yml (the Next
|
||||||
|
Step).
|
||||||
|
- Expand tests beyond the compilation smoke test: unit tests for the
|
||||||
|
extraction, verification, and atomic-publish paths.
|
||||||
|
- Cut a first SemVer release once compliance and test coverage land.
|
||||||
@@ -21,7 +21,11 @@ func CopyFile(src, dst string) (err error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("opening source %s: %w", src, err)
|
return fmt.Errorf("opening source %s: %w", src, err)
|
||||||
}
|
}
|
||||||
defer srcFile.Close()
|
defer func() {
|
||||||
|
if cerr := srcFile.Close(); cerr != nil {
|
||||||
|
slog.Warn("failed to close source file", "src", src, "error", cerr)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
srcInfo, err := srcFile.Stat()
|
srcInfo, err := srcFile.Stat()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -29,7 +29,11 @@ func ExtractDay(srcDBPath, dstDBPath string, targetDay time.Time) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("opening destination database: %w", err)
|
return fmt.Errorf("opening destination database: %w", err)
|
||||||
}
|
}
|
||||||
defer db.Close()
|
defer func() {
|
||||||
|
if cerr := db.Close(); cerr != nil {
|
||||||
|
slog.Warn("failed to close destination database", "path", dstDBPath, "error", cerr)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
// Attach source database
|
// Attach source database
|
||||||
if _, err := db.Exec("ATTACH DATABASE ? AS src", srcDBPath); err != nil {
|
if _, err := db.Exec("ATTACH DATABASE ? AS src", srcDBPath); err != nil {
|
||||||
@@ -42,7 +46,11 @@ func ExtractDay(srcDBPath, dstDBPath string, targetDay time.Time) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("reading source schema: %w", err)
|
return fmt.Errorf("reading source schema: %w", err)
|
||||||
}
|
}
|
||||||
defer rows.Close()
|
defer func() {
|
||||||
|
if cerr := rows.Close(); cerr != nil {
|
||||||
|
slog.Warn("failed to close schema rows", "error", cerr)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
var ddlStatements []string
|
var ddlStatements []string
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
@@ -69,7 +77,9 @@ func ExtractDay(srcDBPath, dstDBPath string, targetDay time.Time) error {
|
|||||||
}
|
}
|
||||||
defer func() {
|
defer func() {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
tx.Rollback()
|
if rerr := tx.Rollback(); rerr != nil && !errors.Is(rerr, sql.ErrTxDone) {
|
||||||
|
slog.Warn("failed to roll back transaction", "error", rerr)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
@@ -132,7 +142,11 @@ func ExtractDay(srcDBPath, dstDBPath string, targetDay time.Time) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("reading source indexes: %w", err)
|
return fmt.Errorf("reading source indexes: %w", err)
|
||||||
}
|
}
|
||||||
defer idxRows.Close()
|
defer func() {
|
||||||
|
if cerr := idxRows.Close(); cerr != nil {
|
||||||
|
slog.Warn("failed to close index rows", "error", cerr)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
var idxStatements []string
|
var idxStatements []string
|
||||||
for idxRows.Next() {
|
for idxRows.Next() {
|
||||||
|
|||||||
@@ -9,6 +9,15 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// cleanup removes a temporary file, logging a warning if removal fails so
|
||||||
|
// that leaked scratch files are surfaced rather than silently ignored. A
|
||||||
|
// missing file is not an error.
|
||||||
|
func cleanup(path string) {
|
||||||
|
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
|
||||||
|
slog.Warn("failed to remove temporary file", "path", path, "error", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func Run(targetDates []time.Time) error {
|
func Run(targetDates []time.Time) error {
|
||||||
snapshotDir, snapshotDate, err := FindLatestDailySnapshot()
|
snapshotDir, snapshotDate, err := FindLatestDailySnapshot()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -100,7 +109,7 @@ func Run(targetDates []time.Time) error {
|
|||||||
if err := ExtractDay(dstDB, extractedDB, targetDay); err != nil {
|
if err := ExtractDay(dstDB, extractedDB, targetDay); err != nil {
|
||||||
if errors.Is(err, ErrNoPosts) {
|
if errors.Is(err, ErrNoPosts) {
|
||||||
slog.Warn("no posts found, skipping day", "date", dayStr)
|
slog.Warn("no posts found, skipping day", "date", dayStr)
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
skipped++
|
skipped++
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -109,7 +118,7 @@ func Run(targetDates []time.Time) error {
|
|||||||
|
|
||||||
// Dump to SQL and compress
|
// Dump to SQL and compress
|
||||||
if err := os.MkdirAll(outputDir, 0755); err != nil {
|
if err := os.MkdirAll(outputDir, 0755); err != nil {
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
return fmt.Errorf("creating output directory %s: %w", outputDir, err)
|
return fmt.Errorf("creating output directory %s: %w", outputDir, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -117,35 +126,35 @@ func Run(targetDates []time.Time) error {
|
|||||||
|
|
||||||
slog.Info("dumping and compressing", "tmp_output", outputTmp)
|
slog.Info("dumping and compressing", "tmp_output", outputTmp)
|
||||||
if err := DumpAndCompress(extractedDB, outputTmp); err != nil {
|
if err := DumpAndCompress(extractedDB, outputTmp); err != nil {
|
||||||
os.Remove(outputTmp)
|
cleanup(outputTmp)
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
return fmt.Errorf("dump and compress for %s: %w", dayStr, err)
|
return fmt.Errorf("dump and compress for %s: %w", dayStr, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Info("verifying compressed output")
|
slog.Info("verifying compressed output")
|
||||||
if err := VerifyOutput(outputTmp); err != nil {
|
if err := VerifyOutput(outputTmp); err != nil {
|
||||||
os.Remove(outputTmp)
|
cleanup(outputTmp)
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
return fmt.Errorf("verification failed for %s: %w", dayStr, err)
|
return fmt.Errorf("verification failed for %s: %w", dayStr, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Atomic rename to final path
|
// Atomic rename to final path
|
||||||
slog.Info("renaming to final output", "from", outputTmp, "to", outputFinal)
|
slog.Info("renaming to final output", "from", outputTmp, "to", outputFinal)
|
||||||
if err := os.Rename(outputTmp, outputFinal); err != nil {
|
if err := os.Rename(outputTmp, outputFinal); err != nil {
|
||||||
os.Remove(outputTmp)
|
cleanup(outputTmp)
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
return fmt.Errorf("atomic rename for %s: %w", dayStr, err)
|
return fmt.Errorf("atomic rename for %s: %w", dayStr, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
info, err := os.Stat(outputFinal)
|
info, err := os.Stat(outputFinal)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
return fmt.Errorf("stat final output: %w", err)
|
return fmt.Errorf("stat final output: %w", err)
|
||||||
}
|
}
|
||||||
slog.Info("day completed", "date", dayStr, "path", outputFinal, "size_bytes", info.Size())
|
slog.Info("day completed", "date", dayStr, "path", outputFinal, "size_bytes", info.Size())
|
||||||
|
|
||||||
// Remove extracted DB to reclaim space immediately
|
// Remove extracted DB to reclaim space immediately
|
||||||
os.Remove(extractedDB)
|
cleanup(extractedDB)
|
||||||
processed++
|
processed++
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,26 @@
|
|||||||
package bsdaily
|
package bsdaily
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
"os"
|
||||||
"os/exec"
|
"os/exec"
|
||||||
"strings"
|
"strings"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// killCat terminates the zstdcat process, ignoring the benign case where it
|
||||||
|
// has already exited (e.g. after receiving SIGPIPE when head closed the pipe)
|
||||||
|
// and logging any other failure.
|
||||||
|
func killCat(cmd *exec.Cmd) {
|
||||||
|
if cmd.Process == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) {
|
||||||
|
slog.Warn("failed to kill zstdcat process", "error", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func VerifyOutput(path string) error {
|
func VerifyOutput(path string) error {
|
||||||
slog.Info("running zstdmt integrity check")
|
slog.Info("running zstdmt integrity check")
|
||||||
testCmd := exec.Command("zstdmt", "--test", path)
|
testCmd := exec.Command("zstdmt", "--test", path)
|
||||||
@@ -34,18 +48,18 @@ func VerifyOutput(path string) error {
|
|||||||
return fmt.Errorf("starting zstdcat: %w", err)
|
return fmt.Errorf("starting zstdcat: %w", err)
|
||||||
}
|
}
|
||||||
if err := headCmd.Start(); err != nil {
|
if err := headCmd.Start(); err != nil {
|
||||||
catCmd.Process.Kill() // Clean up if head fails to start
|
killCat(catCmd) // Clean up if head fails to start
|
||||||
return fmt.Errorf("starting head: %w", err)
|
return fmt.Errorf("starting head: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Wait for head first (it will exit when it has enough lines)
|
// Wait for head first (it will exit when it has enough lines)
|
||||||
if err := headCmd.Wait(); err != nil {
|
if err := headCmd.Wait(); err != nil {
|
||||||
catCmd.Process.Kill()
|
killCat(catCmd)
|
||||||
return fmt.Errorf("head command failed: %w", err)
|
return fmt.Errorf("head command failed: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Kill zstdcat since head closed the pipe (expected SIGPIPE)
|
// Kill zstdcat since head closed the pipe (expected SIGPIPE)
|
||||||
catCmd.Process.Kill()
|
killCat(catCmd)
|
||||||
_ = catCmd.Wait() // Reap the process
|
_ = catCmd.Wait() // Reap the process
|
||||||
|
|
||||||
content := headOut.String()
|
content := headOut.String()
|
||||||
|
|||||||
Reference in New Issue
Block a user