Compare commits

..

1 Commits

Author SHA256 Message Date
Andrii Nikitin
0464324ea7 staging: Add config for poll interval
Some checks failed
go-generate-check / go-generate-check (pull_request) Successful in 44s
Integration tests / t (pull_request) Has been cancelled
Needed for testing
2026-03-05 12:56:03 +01:00
7 changed files with 73 additions and 265 deletions

View File

@@ -23,6 +23,7 @@ import (
"errors"
"fmt"
"io"
"log"
"os"
"slices"
"strings"
@@ -204,21 +205,16 @@ func ReadWorkflowConfig(gitea GiteaFileContentAndRepoFetcher, git_project string
func ResolveWorkflowConfigs(gitea GiteaFileContentAndRepoFetcher, config *ConfigFile) (AutogitConfigs, error) {
configs := make([]*AutogitConfig, 0, len(config.GitProjectNames))
var errs []error
for _, git_project := range config.GitProjectNames {
c, err := ReadWorkflowConfig(gitea, git_project)
if err != nil {
// can't sync, so ignore for now
errs = append(errs, err)
log.Println(err)
} else {
configs = append(configs, c)
}
}
if len(errs) > 0 {
return configs, errors.Join(errs...)
}
return configs, nil
}

View File

@@ -396,17 +396,12 @@ func (e *GitHandlerImpl) GitExecQuietOrPanic(cwd string, params ...string) {
}
type ChanIO struct {
ch chan byte
done chan struct{}
ch chan byte
}
func (c *ChanIO) Write(p []byte) (int, error) {
for _, b := range p {
select {
case c.ch <- b:
case <-c.done:
return 0, io.EOF
}
c.ch <- b
}
return len(p), nil
}
@@ -415,32 +410,21 @@ func (c *ChanIO) Write(p []byte) (int, error) {
func (c *ChanIO) Read(data []byte) (idx int, err error) {
var ok bool
select {
case data[idx], ok = <-c.ch:
data[idx], ok = <-c.ch
if !ok {
err = io.EOF
return
}
idx++
for len(c.ch) > 0 && idx < len(data) {
data[idx], ok = <-c.ch
if !ok {
err = io.EOF
return
}
idx++
case <-c.done:
err = io.EOF
return
}
for len(c.ch) > 0 && idx < len(data) {
select {
case data[idx], ok = <-c.ch:
if !ok {
err = io.EOF
return
}
idx++
case <-c.done:
err = io.EOF
return
default:
return
}
idx++
}
return
@@ -487,14 +471,7 @@ func parseGitMsg(data <-chan byte) (GitMsg, error) {
var size int
pos := 0
for {
c, ok := <-data
if !ok {
return GitMsg{}, io.EOF
}
if c == ' ' {
break
}
for c := <-data; c != ' '; c = <-data {
if (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') {
id[pos] = c
pos++
@@ -506,15 +483,7 @@ func parseGitMsg(data <-chan byte) (GitMsg, error) {
pos = 0
var c byte
for {
var ok bool
c, ok = <-data
if !ok {
return GitMsg{}, io.EOF
}
if c == ' ' || c == '\x00' {
break
}
for c = <-data; c != ' ' && c != '\x00'; c = <-data {
if c >= 'a' && c <= 'z' {
msgType[pos] = c
pos++
@@ -540,14 +509,7 @@ func parseGitMsg(data <-chan byte) (GitMsg, error) {
return GitMsg{}, fmt.Errorf("Invalid object type: '%s'", string(msgType))
}
for {
c, ok := <-data
if !ok {
return GitMsg{}, io.EOF
}
if c == '\x00' {
break
}
for c = <-data; c != '\000'; c = <-data {
if c >= '0' && c <= '9' {
size = size*10 + (int(c) - '0')
} else {
@@ -566,37 +528,18 @@ func parseGitCommitHdr(oldHdr [2]string, data <-chan byte) ([2]string, int, erro
hdr := make([]byte, 0, 60)
val := make([]byte, 0, 1000)
c, ok := <-data
if !ok {
return [2]string{}, 0, io.EOF
}
c := <-data
size := 1
if c != '\n' { // end of header marker
for {
if c == ' ' {
break
}
for ; c != ' '; c = <-data {
hdr = append(hdr, c)
size++
var ok bool
c, ok = <-data
if !ok {
return [2]string{}, size, io.EOF
}
}
if size == 1 { // continuation header here
hdr = []byte(oldHdr[0])
val = append([]byte(oldHdr[1]), '\n')
}
for {
var ok bool
c, ok = <-data
if !ok {
return [2]string{}, size, io.EOF
}
if c == '\n' {
break
}
for c := <-data; c != '\n'; c = <-data {
val = append(val, c)
size++
}
@@ -609,14 +552,7 @@ func parseGitCommitHdr(oldHdr [2]string, data <-chan byte) ([2]string, int, erro
func parseGitCommitMsg(data <-chan byte, l int) (string, error) {
msg := make([]byte, 0, l)
for {
c, ok := <-data
if !ok {
return string(msg), io.EOF
}
if c == '\x00' {
break
}
for c := <-data; c != '\x00'; c = <-data {
msg = append(msg, c)
l--
}
@@ -642,7 +578,7 @@ func parseGitCommit(data <-chan byte) (GitCommit, error) {
var hdr [2]string
hdr, size, err := parseGitCommitHdr(hdr, data)
if err != nil {
return GitCommit{}, err
return GitCommit{}, nil
}
l -= size
@@ -663,28 +599,14 @@ func parseGitCommit(data <-chan byte) (GitCommit, error) {
func parseTreeEntry(data <-chan byte, hashLen int) (GitTreeEntry, error) {
var e GitTreeEntry
for {
c, ok := <-data
if !ok {
return e, io.EOF
}
if c == ' ' {
break
}
for c := <-data; c != ' '; c = <-data {
e.mode = e.mode*8 + int(c-'0')
e.size++
}
e.size++
name := make([]byte, 0, 128)
for {
c, ok := <-data
if !ok {
return e, io.EOF
}
if c == '\x00' {
break
}
for c := <-data; c != '\x00'; c = <-data {
name = append(name, c)
e.size++
}
@@ -695,10 +617,7 @@ func parseTreeEntry(data <-chan byte, hashLen int) (GitTreeEntry, error) {
hash := make([]byte, 0, hashLen*2)
for range hashLen {
c, ok := <-data
if !ok {
return e, io.EOF
}
c := <-data
hash = append(hash, hexBinToAscii[((c&0xF0)>>4)], hexBinToAscii[c&0xF])
}
e.hash = string(hash)
@@ -719,16 +638,13 @@ func parseGitTree(data <-chan byte) (GitTree, error) {
for parsedLen < hdr.size {
entry, err := parseTreeEntry(data, len(hdr.hash)/2)
if err != nil {
return GitTree{}, err
return GitTree{}, nil
}
t.items = append(t.items, entry)
parsedLen += entry.size
}
c, ok := <-data // \0 read
if !ok {
return t, io.EOF
}
c := <-data // \0 read
if c != '\x00' {
return t, fmt.Errorf("Unexpected character during git tree data read")
@@ -749,16 +665,9 @@ func parseGitBlob(data <-chan byte) ([]byte, error) {
d := make([]byte, hdr.size)
for l := 0; l < hdr.size; l++ {
var ok bool
d[l], ok = <-data
if !ok {
return d, io.EOF
}
}
eob, ok := <-data
if !ok {
return d, io.EOF
d[l] = <-data
}
eob := <-data
if eob != '\x00' {
return d, fmt.Errorf("invalid byte read in parseGitBlob")
}
@@ -770,25 +679,16 @@ func (e *GitHandlerImpl) GitParseCommits(cwd string, commitIDs []string) (parsed
var done sync.Mutex
done.Lock()
done_signal := make(chan struct{})
var once sync.Once
close_done := func() {
once.Do(func() {
close(done_signal)
})
}
data_in, data_out := ChanIO{make(chan byte), done_signal}, ChanIO{make(chan byte), done_signal}
data_in, data_out := ChanIO{make(chan byte)}, ChanIO{make(chan byte)}
parsedCommits = make([]GitCommit, 0, len(commitIDs))
go func() {
defer done.Unlock()
defer close_done()
defer close(data_out.ch)
for _, id := range commitIDs {
data_out.Write([]byte(id))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
c, e := parseGitCommit(data_in.ch)
if e != nil {
err = fmt.Errorf("Error parsing git commit: %w", e)
@@ -815,14 +715,12 @@ func (e *GitHandlerImpl) GitParseCommits(cwd string, commitIDs []string) (parsed
LogDebug("command run:", cmd.Args)
if e := cmd.Run(); e != nil {
LogError(e)
close_done()
close(data_in.ch)
close(data_out.ch)
return nil, e
}
done.Lock()
close_done()
close(data_in.ch)
return
}
@@ -831,21 +729,15 @@ func (e *GitHandlerImpl) GitCatFile(cwd, commitId, filename string) (data []byte
var done sync.Mutex
done.Lock()
done_signal := make(chan struct{})
var once sync.Once
close_done := func() {
once.Do(func() {
close(done_signal)
})
}
data_in, data_out := ChanIO{make(chan byte), done_signal}, ChanIO{make(chan byte), done_signal}
data_in, data_out := ChanIO{make(chan byte)}, ChanIO{make(chan byte)}
go func() {
defer done.Unlock()
defer close_done()
defer close(data_out.ch)
data_out.Write([]byte(commitId))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var c GitCommit
c, err = parseGitCommit(data_in.ch)
if err != nil {
@@ -853,9 +745,11 @@ func (e *GitHandlerImpl) GitCatFile(cwd, commitId, filename string) (data []byte
return
}
data_out.Write([]byte(c.Tree))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var tree GitTree
tree, err = parseGitTree(data_in.ch)
if err != nil {
LogError("Error parsing git tree:", err)
return
@@ -865,7 +759,7 @@ func (e *GitHandlerImpl) GitCatFile(cwd, commitId, filename string) (data []byte
if te.isBlob() && te.name == filename {
LogInfo("blob", te.hash)
data_out.Write([]byte(te.hash))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
data, err = parseGitBlob(data_in.ch)
return
}
@@ -890,13 +784,11 @@ func (e *GitHandlerImpl) GitCatFile(cwd, commitId, filename string) (data []byte
LogDebug("command run:", cmd.Args)
if e := cmd.Run(); e != nil {
LogError(e)
close_done()
close(data_in.ch)
close(data_out.ch)
return nil, e
}
done.Lock()
close_done()
close(data_in.ch)
return
}
@@ -906,24 +798,16 @@ func (e *GitHandlerImpl) GitDirectoryList(gitPath, commitId string) (directoryLi
directoryList = make(map[string]string)
done.Lock()
done_signal := make(chan struct{})
var once sync.Once
close_done := func() {
once.Do(func() {
close(done_signal)
})
}
data_in, data_out := ChanIO{make(chan byte), done_signal}, ChanIO{make(chan byte), done_signal}
data_in, data_out := ChanIO{make(chan byte)}, ChanIO{make(chan byte)}
LogDebug("Getting directory for:", commitId)
go func() {
defer done.Unlock()
defer close_done()
defer close(data_out.ch)
data_out.Write([]byte(commitId))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var c GitCommit
c, err = parseGitCommit(data_in.ch)
if err != nil {
@@ -939,7 +823,7 @@ func (e *GitHandlerImpl) GitDirectoryList(gitPath, commitId string) (directoryLi
delete(trees, p)
data_out.Write([]byte(tree))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var tree GitTree
tree, err = parseGitTree(data_in.ch)
@@ -973,14 +857,12 @@ func (e *GitHandlerImpl) GitDirectoryList(gitPath, commitId string) (directoryLi
LogDebug("command run:", cmd.Args)
if e := cmd.Run(); e != nil {
LogError(e)
close_done()
close(data_in.ch)
close(data_out.ch)
return directoryList, e
}
done.Lock()
close_done()
close(data_in.ch)
return directoryList, err
}
@@ -990,14 +872,7 @@ func (e *GitHandlerImpl) GitDirectoryContentList(gitPath, commitId string) (dire
directoryList = make(map[string]string)
done.Lock()
done_signal := make(chan struct{})
var once sync.Once
close_done := func() {
once.Do(func() {
close(done_signal)
})
}
data_in, data_out := ChanIO{make(chan byte), done_signal}, ChanIO{make(chan byte), done_signal}
data_in, data_out := ChanIO{make(chan byte)}, ChanIO{make(chan byte)}
LogDebug("Getting directory content for:", commitId)
@@ -1006,7 +881,7 @@ func (e *GitHandlerImpl) GitDirectoryContentList(gitPath, commitId string) (dire
defer close(data_out.ch)
data_out.Write([]byte(commitId))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var c GitCommit
c, err = parseGitCommit(data_in.ch)
if err != nil {
@@ -1022,7 +897,7 @@ func (e *GitHandlerImpl) GitDirectoryContentList(gitPath, commitId string) (dire
delete(trees, p)
data_out.Write([]byte(tree))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var tree GitTree
tree, err = parseGitTree(data_in.ch)
@@ -1058,14 +933,12 @@ func (e *GitHandlerImpl) GitDirectoryContentList(gitPath, commitId string) (dire
LogDebug("command run:", cmd.Args)
if e := cmd.Run(); e != nil {
LogError(e)
close_done()
close(data_in.ch)
close(data_out.ch)
return directoryList, e
}
done.Lock()
close_done()
close(data_in.ch)
return directoryList, err
}
@@ -1075,24 +948,16 @@ func (e *GitHandlerImpl) GitSubmoduleList(gitPath, commitId string) (submoduleLi
submoduleList = make(map[string]string)
done.Lock()
done_signal := make(chan struct{})
var once sync.Once
close_done := func() {
once.Do(func() {
close(done_signal)
})
}
data_in, data_out := ChanIO{make(chan byte), done_signal}, ChanIO{make(chan byte), done_signal}
data_in, data_out := ChanIO{make(chan byte)}, ChanIO{make(chan byte)}
LogDebug("Getting submodules for:", commitId)
go func() {
defer done.Unlock()
defer close_done()
defer close(data_out.ch)
data_out.Write([]byte(commitId))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var c GitCommit
c, err = parseGitCommit(data_in.ch)
if err != nil {
@@ -1108,7 +973,7 @@ func (e *GitHandlerImpl) GitSubmoduleList(gitPath, commitId string) (submoduleLi
delete(trees, p)
data_out.Write([]byte(tree))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
var tree GitTree
tree, err = parseGitTree(data_in.ch)
@@ -1145,26 +1010,17 @@ func (e *GitHandlerImpl) GitSubmoduleList(gitPath, commitId string) (submoduleLi
LogDebug("command run:", cmd.Args)
if e := cmd.Run(); e != nil {
LogError(e)
close_done()
close(data_in.ch)
close(data_out.ch)
return submoduleList, e
}
done.Lock()
close_done()
close(data_in.ch)
return submoduleList, err
}
func (e *GitHandlerImpl) GitSubmoduleCommitId(cwd, packageName, commitId string) (subCommitId string, valid bool) {
done_signal := make(chan struct{})
var once sync.Once
close_done := func() {
once.Do(func() {
close(done_signal)
})
}
data_in, data_out := ChanIO{make(chan byte), done_signal}, ChanIO{make(chan byte), done_signal}
data_in, data_out := ChanIO{make(chan byte)}, ChanIO{make(chan byte)}
var wg sync.WaitGroup
wg.Add(1)
@@ -1180,18 +1036,17 @@ func (e *GitHandlerImpl) GitSubmoduleCommitId(cwd, packageName, commitId string)
}()
defer wg.Done()
defer close_done()
defer close(data_out.ch)
data_out.Write([]byte(commitId))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
c, err := parseGitCommit(data_in.ch)
if err != nil {
LogError("Error parsing git commit:", err)
panic(err)
}
data_out.Write([]byte(c.Tree))
data_out.Write([]byte{0})
data_out.ch <- '\x00'
tree, err := parseGitTree(data_in.ch)
if err != nil {
@@ -1223,14 +1078,12 @@ func (e *GitHandlerImpl) GitSubmoduleCommitId(cwd, packageName, commitId string)
LogDebug("command run:", cmd.Args)
if e := cmd.Run(); e != nil {
LogError(e)
close_done()
close(data_in.ch)
close(data_out.ch)
return subCommitId, false
}
wg.Wait()
close_done()
close(data_in.ch)
return subCommitId, len(subCommitId) > 0
}

View File

@@ -28,7 +28,6 @@ import (
"slices"
"strings"
"testing"
"time"
)
func TestGitClone(t *testing.T) {
@@ -718,44 +717,3 @@ func TestGitDirectoryListRepro(t *testing.T) {
t.Errorf("Expected 'subdir' in directory list, got %v", dirs)
}
}
func TestGitDeadlockFix(t *testing.T) {
gitDir := t.TempDir()
testDir, _ := os.Getwd()
cmd := exec.Command("/usr/bin/bash", path.Join(testDir, "tsetup.sh"))
cmd.Dir = gitDir
_, err := cmd.CombinedOutput()
gh, err := AllocateGitWorkTree(gitDir, "Test", "test@example.com")
if err != nil {
t.Fatal(err)
}
h, err := gh.ReadExistingPath(".")
if err != nil {
t.Fatal(err)
}
defer h.Close()
// Use a blob ID to trigger error in GitParseCommits
// This ensures that the function returns error immediately and doesn't deadlock
blobId := "81aba862107f1e2f5312e165453955485f424612f313d6c2fb1b31fef9f82a14"
done := make(chan error)
go func() {
_, err := h.GitParseCommits("", []string{blobId})
done <- err
}()
select {
case err := <-done:
if err == nil {
t.Error("Expected error from GitParseCommits with blob ID, got nil")
} else {
// This is expected
t.Logf("Got expected error: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("GitParseCommits deadlocked! Fix is NOT working.")
}
}

View File

@@ -327,7 +327,6 @@ func main() {
interval := flag.Int64("interval", 10, "Notification polling interval in minutes (min 1 min)")
configFile := flag.String("config", "", "PrjGit listing config file")
logging := flag.String("logging", "info", "Logging level: [none, error, info, debug]")
exitOnConfigError := flag.Bool("exit-on-config-error", false, "Exit if any repository in configuration cannot be resolved")
flag.BoolVar(&common.IsDryRun, "dry", false, "Dry run, no effect. For debugging")
flag.Parse()
@@ -383,10 +382,8 @@ func main() {
giteaTransport := common.AllocateGiteaTransport(*giteaUrl)
configs, err := common.ResolveWorkflowConfigs(giteaTransport, configData)
if err != nil {
common.LogError("Failed to resolve some configuration repositories:", err)
if *exitOnConfigError {
return
}
common.LogError("Cannot parse workflow configs:", err)
return
}
reviewer, err := giteaTransport.GetCurrentUser()

View File

@@ -1171,6 +1171,7 @@ var IsDryRun bool
var ProcessPROnly string
var ObsClient common.ObsClientInterface
var BotUser string
var PollInterval = 5 * time.Minute
func ObsWebHostFromApiHost(apihost string) string {
u, err := url.Parse(apihost)
@@ -1193,9 +1194,18 @@ func main() {
flag.StringVar(&ObsApiHost, "obs", "", "API for OBS instance")
flag.StringVar(&ObsWebHost, "obs-web", "", "Web OBS instance, if not derived from the obs config")
flag.BoolVar(&IsDryRun, "dry", false, "Dry-run, don't actually create any build projects or review changes")
pollIntervalStr := flag.String("poll-interval", common.GetEnvOverrideString(os.Getenv("AUTOGITS_STAGING_BOT_POLL_INTERVAL"), ""), "Polling interval for notifications (e.g. 5m, 10s)")
debug := flag.Bool("debug", false, "Turns on debug logging")
flag.Parse()
if len(*pollIntervalStr) > 0 {
if d, err := time.ParseDuration(*pollIntervalStr); err == nil {
PollInterval = d
} else {
common.LogError("Invalid poll interval:", err)
}
}
if *debug {
common.SetLoggingLevel(common.LogLevelDebug)
} else {
@@ -1264,6 +1274,6 @@ func main() {
for {
PollWorkNotifications(ObsClient, gitea)
common.LogInfo("Poll cycle finished")
time.Sleep(5 * time.Minute)
time.Sleep(PollInterval)
}
}

View File

@@ -503,10 +503,7 @@ func updateConfiguration(configFilename string, orgs *[]string) {
os.Exit(4)
}
configs, err := common.ResolveWorkflowConfigs(gitea, configFile)
if err != nil {
common.LogError("Failed to resolve some configuration repositories:", err)
}
configs, _ := common.ResolveWorkflowConfigs(gitea, configFile)
configuredRepos = make(map[string][]*common.AutogitConfig)
*orgs = make([]string, 0, 1)
for _, c := range configs {

View File

@@ -58,7 +58,6 @@ func main() {
checkOnStart := flag.Bool("check-on-start", common.GetEnvOverrideBool(os.Getenv("AUTOGITS_CHECK_ON_START"), false), "Check all repositories for consistency on start, without delays")
checkIntervalHours := flag.Float64("check-interval", 5, "Check interval (+-random delay) for repositories for consitency, in hours")
flag.BoolVar(&ListPROnly, "list-prs-only", false, "Only lists PRs without acting on them")
exitOnConfigError := flag.Bool("exit-on-config-error", false, "Exit if any repository in configuration cannot be resolved")
flag.Int64Var(&PRID, "id", -1, "Process only the specific ID and ignore the rest. Use for debugging")
basePath := flag.String("repo-path", common.GetEnvOverrideString(os.Getenv("AUTOGITS_REPO_PATH"), ""), "Repository path. Default is temporary directory")
pr := flag.String("only-pr", "", "Only specific PR to process. For debugging")
@@ -98,10 +97,8 @@ func main() {
configs, err := common.ResolveWorkflowConfigs(Gitea, config)
if err != nil {
common.LogError("Failed to resolve some configuration repositories:", err)
if *exitOnConfigError {
return
}
common.LogError("Cannot resolve config files:", err)
return
}
for _, c := range configs {