mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-08 23:54:39 +00:00
fix(cli): second audit pass — read routing, status perf, publish-ack + robustness
Findings from a second review round (two independent reviewers), with fixes: Read path (git issues/patches/prs/thread): - **Reads ignored the repo's own relays** (correctness). They queried only the account outbox/bootstrap (general relays); NIP-34 events live on the repo announcement's advertised relays (often GRASP/git-specific), which general relays don't mirror — so `amy git issues <repo>` with no --relay could return empty. Now fetch the announcement once and read from queryTargets ∪ its advertised `relays`. Verified live: `git issues`/`git prs` on the amethyst repo now return real events (and derive `closed`) with NO --relay. - **O(items × statuses) status rescan** with un-memoized `rootEventId()` reparse → pre-group statuses by root id once (O(1) lookup per item). - **Status query could truncate / exceed relay caps**: statuses are now paged (`drainAllPages`) and the `#e` id set is chunked to 50 (under the common ~100-value relay filter cap). - **Latency regression**: capped the list `drainAllPages` idle timeout to 12s (was the 30s default; `drain` had been 8s). - **Nondeterministic status on same-second ties** → deterministic id tie-break. - Reuse the fetched repo for the maintainer set (removes a redundant round-trip). Write path: - **`git init` silently reported success when the 30618 state publish failed** — its ack was dropped. Now surfaced as `state_published_to`/`state_rejected_by` with a stderr warning on total rejection. - **`git apply`** feeds stdin as UTF-8 (was JVM default charset — corrupted non-ASCII patches) and joins the stdin thread in `finally` (no leak on error). - **`normalizeCloneUrl`** drops the port from `ssh://git@host:port/…` (it was carried into the https URL, making it unreachable). - **Delivery fallback** to the account outbox (repo unresolved / no advertised relays) now warns to stderr instead of reporting silent success. Known limitation (documented, not fixed): patch-revision-chain status derivation follows only the root item, and `git thread` shows first-level replies only (nested trees and 1619 PR-updates are out of scope). 38/38 harness green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UKMaNoK5M2PQKCAxhxWzPr
This commit is contained in:
Claude
@@ -99,8 +99,9 @@ object GitApplyCommand {
|
||||
repoDir: File,
|
||||
input: String?,
|
||||
vararg gitArgs: String,
|
||||
): Pair<Int, String> =
|
||||
try {
|
||||
): Pair<Int, String> {
|
||||
var writer: Thread? = null
|
||||
return try {
|
||||
val proc =
|
||||
ProcessBuilder(listOf("git", *gitArgs))
|
||||
.directory(repoDir)
|
||||
@@ -109,18 +110,23 @@ object GitApplyCommand {
|
||||
// Feed stdin on a separate thread while we drain stdout on this one: a
|
||||
// large patch (bigger than the OS pipe buffer) would otherwise deadlock
|
||||
// — git blocks writing output we haven't read, we block writing stdin.
|
||||
val writer =
|
||||
// The patch was decoded as UTF-8, so write it back as UTF-8 (not the JVM
|
||||
// default charset, which would corrupt non-ASCII filenames/messages).
|
||||
writer =
|
||||
if (input != null) {
|
||||
thread(name = "git-stdin") { runCatching { proc.outputStream.use { it.write(input.toByteArray()) } } }
|
||||
thread(name = "git-stdin") { runCatching { proc.outputStream.use { it.write(input.toByteArray(Charsets.UTF_8)) } } }
|
||||
} else {
|
||||
proc.outputStream.close()
|
||||
null
|
||||
}
|
||||
val out = proc.inputStream.readBytes().decodeToString()
|
||||
val code = proc.waitFor()
|
||||
writer?.join()
|
||||
code to out
|
||||
proc.waitFor() to out
|
||||
} catch (e: Exception) {
|
||||
1 to (e.message ?: "could not run git (is it installed and is this a git repo?)")
|
||||
} finally {
|
||||
// Always join so a non-daemon stdin thread can't outlive the call (e.g.
|
||||
// under the in-process test harness, which doesn't exitProcess).
|
||||
writer?.join()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -119,11 +119,17 @@ object GitInitCommand {
|
||||
if (refs.isNotEmpty() || head != null) {
|
||||
val stateTemplate = GitRepositoryStateEvent.build(dTag = identifier, refs = refs, head = head)
|
||||
val signedState = ctx.signer.sign(stateTemplate)
|
||||
ctx.publish(signedState, targets)
|
||||
// Surface the state publish result separately (state_*) rather than
|
||||
// dropping it — otherwise a fully-rejected 30618 reads as success.
|
||||
val stateAck = ctx.publish(signedState, targets)
|
||||
result["state_event_id"] = signedState.id
|
||||
result["branches"] = refs.count { it.kind == RefTag.Kind.BRANCH }
|
||||
result["tags"] = refs.count { it.kind == RefTag.Kind.TAG }
|
||||
result["head"] = head
|
||||
result.putAll(RawEventSupport.ackFields(stateAck).mapKeys { "state_${it.key}" })
|
||||
if (stateAck.isNotEmpty() && stateAck.none { it.value.accepted }) {
|
||||
System.err.println("[git init] warning: no relay accepted the repository-state (30618) event — branches/tags/HEAD were not delivered.")
|
||||
}
|
||||
}
|
||||
}
|
||||
Output.emit(result + RawEventSupport.ackFields(ackAnnounce))
|
||||
@@ -180,7 +186,15 @@ object GitInitCommand {
|
||||
val path = rest.substringAfter(':')
|
||||
"https://$host/$path"
|
||||
}
|
||||
url.startsWith("ssh://git@") -> "https://" + url.removePrefix("ssh://git@")
|
||||
url.startsWith("ssh://git@") -> {
|
||||
// ssh://git@host[:port]/owner/repo.git → https://host/owner/repo.git
|
||||
// (drop the SSH port; carrying it into the https URL makes it unreachable).
|
||||
val rest = url.removePrefix("ssh://git@")
|
||||
val slash = rest.indexOf('/')
|
||||
val hostPort = if (slash >= 0) rest.take(slash) else rest
|
||||
val path = if (slash >= 0) rest.substring(slash) else ""
|
||||
"https://${hostPort.substringBefore(':')}$path"
|
||||
}
|
||||
else -> url
|
||||
}
|
||||
|
||||
|
||||
@@ -26,12 +26,15 @@ import com.vitorpamplona.amethyst.cli.DataDir
|
||||
import com.vitorpamplona.amethyst.cli.Output
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Address
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip22Comments.CommentEvent
|
||||
import com.vitorpamplona.quartz.nip34Git.issue.GitIssueEvent
|
||||
import com.vitorpamplona.quartz.nip34Git.patch.GitPatchEvent
|
||||
import com.vitorpamplona.quartz.nip34Git.pr.GitPullRequestEvent
|
||||
import com.vitorpamplona.quartz.nip34Git.reply.GitReplyEvent
|
||||
import com.vitorpamplona.quartz.nip34Git.repository.GitRepositoryEvent
|
||||
import com.vitorpamplona.quartz.nip34Git.status.GitStatusEvent
|
||||
|
||||
/**
|
||||
@@ -48,6 +51,12 @@ object GitReadCommands {
|
||||
GitStatusEvent.KIND_DRAFT,
|
||||
)
|
||||
|
||||
/** Idle timeout for the list reads — tighter than `drainAllPages`' 30s default so a dead relay can't stall a list. */
|
||||
private const val READ_TIMEOUT_MS = 12_000L
|
||||
|
||||
/** Max event ids per `#e` status filter — many relays cap tag-filter values around 100, so stay well under. */
|
||||
private const val STATUS_ID_CHUNK = 50
|
||||
|
||||
suspend fun issues(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
@@ -81,16 +90,21 @@ object GitReadCommands {
|
||||
Context.openOrAnonymous(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
val repoAddress = GitSupport.repoCoordinate(addr)
|
||||
val relays = RawEventSupport.queryTargets(ctx, args)
|
||||
// Fetch the announcement once: it supplies both the maintainer set
|
||||
// (status authority) AND the advertised relays the events actually live
|
||||
// on. Reading only from general relays misses GRASP-hosted repos.
|
||||
val repo = GitSupport.fetchRepo(ctx, addr, args)
|
||||
val relays = GitSupport.readTargets(ctx, repo, args)
|
||||
|
||||
// Two queries so item volume and status volume never starve each other
|
||||
// (a busy repo can have far more status events than items). First page
|
||||
// the items to the limit; then fetch exactly the status events that
|
||||
// `e`-reference those items, so derived status is never truncated away.
|
||||
// Two queries so item volume and status volume never starve each other.
|
||||
// Items are paged to the limit; statuses are fetched separately by the
|
||||
// items' ids so derived status is never truncated by item volume.
|
||||
val itemEvents =
|
||||
ctx
|
||||
.drainAllPages(relays.associateWith { listOf(Filter(kinds = listOf(itemKind), tags = mapOf("a" to listOf(repoAddress)), limit = limit)) })
|
||||
.asSequence()
|
||||
.drainAllPages(
|
||||
relays.associateWith { listOf(Filter(kinds = listOf(itemKind), tags = mapOf("a" to listOf(repoAddress)), limit = limit)) },
|
||||
timeoutMs = READ_TIMEOUT_MS,
|
||||
).asSequence()
|
||||
.map { it.second }
|
||||
.filter { it.kind == itemKind }
|
||||
.distinctBy { it.id }
|
||||
@@ -98,22 +112,12 @@ object GitReadCommands {
|
||||
.take(limit)
|
||||
.toList()
|
||||
|
||||
val itemIds = itemEvents.map { it.id }
|
||||
val statuses =
|
||||
if (itemIds.isEmpty()) {
|
||||
emptyList()
|
||||
} else {
|
||||
ctx
|
||||
.drain(relays.associateWith { listOf(Filter(kinds = STATUS_KINDS, tags = mapOf("e" to itemIds))) })
|
||||
.map { it.second }
|
||||
.filterIsInstance<GitStatusEvent>()
|
||||
.distinctBy { it.id }
|
||||
}
|
||||
val authorities = repoAuthorities(ctx, addr, args)
|
||||
val statusesByRoot = fetchStatusesFor(ctx, relays, itemEvents.map { it.id }).groupBy { it.rootEventId() }
|
||||
val authorities = repoAuthorities(repo, addr)
|
||||
|
||||
val items =
|
||||
itemEvents
|
||||
.map { item -> GitSupport.targetSummary(item) + mapOf("status" to latestStatus(item, statuses, authorities)) }
|
||||
.map { item -> GitSupport.targetSummary(item) + mapOf("status" to latestStatus(item, statusesByRoot, authorities)) }
|
||||
.filter { wanted == null || it["status"] in wanted }
|
||||
|
||||
Output.emit(mapOf("repository" to repoAddress, "count" to items.size, "items" to items))
|
||||
@@ -124,6 +128,10 @@ object GitReadCommands {
|
||||
/**
|
||||
* `amy git thread TARGET` — the target event plus its status timeline and
|
||||
* NIP-22 comments (and legacy kind:1622 replies).
|
||||
*
|
||||
* Scope: first-level replies/statuses that `e`-reference the target. Nested
|
||||
* comment trees and PR-update (1619) events (which use NIP-22 uppercase `E`)
|
||||
* are out of scope here.
|
||||
*/
|
||||
suspend fun thread(
|
||||
dataDir: DataDir,
|
||||
@@ -141,7 +149,9 @@ object GitReadCommands {
|
||||
val target =
|
||||
GitSupport.fetchEvent(ctx, id, args)
|
||||
?: return Output.error("not_found", "no event found for $ref")
|
||||
val relays = RawEventSupport.queryTargets(ctx, args)
|
||||
val repoATag = GitSupport.repositoryOf(target)
|
||||
val repo = repoATag?.let { GitSupport.fetchRepo(ctx, Address(it.kind, it.pubKeyHex, it.dTag), args) }
|
||||
val relays = GitSupport.readTargets(ctx, repo, args)
|
||||
// Everything that `e`-references the target: statuses, comments, replies.
|
||||
val related =
|
||||
ctx
|
||||
@@ -149,12 +159,13 @@ object GitReadCommands {
|
||||
relays.associateWith {
|
||||
listOf(Filter(kinds = STATUS_KINDS + listOf(CommentEvent.KIND, GitReplyEvent.KIND), tags = mapOf("e" to listOf(id))))
|
||||
},
|
||||
timeoutMs = READ_TIMEOUT_MS,
|
||||
).map { it.second }
|
||||
.distinctBy { it.id }
|
||||
|
||||
val repoATag = GitSupport.repositoryOf(target)
|
||||
val authorities = repoATag?.let { repoAuthorities(ctx, Address(it.kind, it.pubKeyHex, it.dTag), args) } ?: setOf(target.pubKey)
|
||||
val authorities = repoATag?.let { repoAuthorities(repo, Address(it.kind, it.pubKeyHex, it.dTag)) } ?: setOf(target.pubKey)
|
||||
val statuses = related.filterIsInstance<GitStatusEvent>().filter { it.rootEventId() == id }
|
||||
val statusesByRoot = statuses.groupBy { it.rootEventId() }
|
||||
val comments =
|
||||
related
|
||||
.filter { it.kind == CommentEvent.KIND || it.kind == GitReplyEvent.KIND }
|
||||
@@ -165,7 +176,7 @@ object GitReadCommands {
|
||||
GitSupport.targetSummary(target) +
|
||||
mapOf(
|
||||
"content" to target.content,
|
||||
"status" to latestStatus(target, statuses, authorities),
|
||||
"status" to latestStatus(target, statusesByRoot, authorities),
|
||||
"status_events" to
|
||||
statuses.sortedBy { it.createdAt }.map {
|
||||
mapOf("event_id" to it.id, "status" to GitSupport.statusLabel(it.kind), "author" to it.pubKey, "created_at" to it.createdAt)
|
||||
@@ -179,34 +190,56 @@ object GitReadCommands {
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/** The set of pubkeys whose status is authoritative for a repo: the owner + declared maintainers. */
|
||||
private suspend fun repoAuthorities(
|
||||
/**
|
||||
* Fetch the status events that `e`-reference [itemIds]. Chunked to stay under
|
||||
* relay tag-filter value caps, and paged (`drainAllPages`) so a heavily
|
||||
* reopened/closed item's older status can't fall off a single page.
|
||||
*/
|
||||
private suspend fun fetchStatusesFor(
|
||||
ctx: Context,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
itemIds: List<HexKey>,
|
||||
): List<GitStatusEvent> {
|
||||
if (itemIds.isEmpty()) return emptyList()
|
||||
return itemIds
|
||||
.chunked(STATUS_ID_CHUNK)
|
||||
.flatMap { chunk ->
|
||||
ctx
|
||||
.drainAllPages(
|
||||
relays.associateWith { listOf(Filter(kinds = STATUS_KINDS, tags = mapOf("e" to chunk))) },
|
||||
timeoutMs = READ_TIMEOUT_MS,
|
||||
).map { it.second }
|
||||
}.filterIsInstance<GitStatusEvent>()
|
||||
.distinctBy { it.id }
|
||||
}
|
||||
|
||||
/** The set of pubkeys whose status is authoritative for a repo: the owner + declared maintainers. */
|
||||
private fun repoAuthorities(
|
||||
repo: GitRepositoryEvent?,
|
||||
addr: Address,
|
||||
args: Args,
|
||||
): Set<String> {
|
||||
val repo = GitSupport.fetchRepo(ctx, addr, args)
|
||||
return buildSet {
|
||||
): Set<HexKey> =
|
||||
buildSet {
|
||||
add(addr.pubKeyHex)
|
||||
repo?.maintainers()?.let { addAll(it) }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The authoritative status label for [item]: the newest status event (by
|
||||
* `created_at`) that `e`-roots this item and is signed by the item author,
|
||||
* the repo owner, or a maintainer. Defaults to `open` when none exists.
|
||||
* `created_at`, id as a deterministic tie-break) that `e`-roots this item and
|
||||
* is signed by the item author, the repo owner, or a maintainer. Defaults to
|
||||
* `open` when none exists. [statusesByRoot] is pre-grouped by root event id so
|
||||
* this is an O(1) lookup instead of an O(items × statuses) rescan.
|
||||
*/
|
||||
private fun latestStatus(
|
||||
item: Event,
|
||||
statuses: List<GitStatusEvent>,
|
||||
authorities: Set<String>,
|
||||
statusesByRoot: Map<HexKey?, List<GitStatusEvent>>,
|
||||
authorities: Set<HexKey>,
|
||||
): String {
|
||||
val allowed = authorities + item.pubKey
|
||||
val newest =
|
||||
statuses
|
||||
.filter { it.rootEventId() == item.id && it.pubKey in allowed }
|
||||
.maxByOrNull { it.createdAt }
|
||||
statusesByRoot[item.id]
|
||||
?.filter { it.pubKey in allowed }
|
||||
?.maxWithOrNull(compareBy({ it.createdAt }, { it.id }))
|
||||
return GitSupport.statusLabel(newest?.kind)
|
||||
}
|
||||
|
||||
|
||||
@@ -129,13 +129,44 @@ object GitSupport {
|
||||
repo: GitRepositoryEvent?,
|
||||
args: Args,
|
||||
): Set<NormalizedRelayUrl> {
|
||||
val flag = RawEventSupport.relayFlag(args)
|
||||
if (flag.isNotEmpty()) return flag
|
||||
val advertised =
|
||||
repo
|
||||
?.relays()
|
||||
?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }
|
||||
?.toSet()
|
||||
.orEmpty()
|
||||
return RawEventSupport.relayFlag(args).ifEmpty { advertised }.ifEmpty { ctx.outboxRelays() }
|
||||
if (advertised.isNotEmpty()) return advertised
|
||||
// Falling back to the account outbox: the repo couldn't be resolved or
|
||||
// advertises no relays, so a maintainer watching only the repo's NIP-34
|
||||
// relays may never see this event. Say so instead of reporting silent success.
|
||||
System.err.println(
|
||||
"[git] warning: no repository relays known — delivering to your outbox. " +
|
||||
"A maintainer watching only the repo's relays may not see this; pass --relay to target them.",
|
||||
)
|
||||
return ctx.outboxRelays()
|
||||
}
|
||||
|
||||
/**
|
||||
* Where to READ a repo's collaboration events from: the account's query
|
||||
* relays (explicit `--relay`, else outbox, else bootstrap) UNION the repo
|
||||
* announcement's own advertised `relays` — the NIP-34 monitored set where
|
||||
* patches/issues/statuses actually live. Without the union, a repo hosted on
|
||||
* GRASP/git-specific relays (which general relays don't mirror) reads empty.
|
||||
*/
|
||||
suspend fun readTargets(
|
||||
ctx: Context,
|
||||
repo: GitRepositoryEvent?,
|
||||
args: Args,
|
||||
): Set<NormalizedRelayUrl> {
|
||||
val advertised =
|
||||
repo
|
||||
?.relays()
|
||||
?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }
|
||||
?.toSet()
|
||||
.orEmpty()
|
||||
return RawEventSupport.queryTargets(ctx, args) + advertised
|
||||
}
|
||||
|
||||
/** Parse a `--flag a,b,c` CSV flag into a trimmed, non-empty list. */
|
||||
|
||||
Reference in New Issue
Block a user