laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-07-30 19:46:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(quartz,cli): audit fixes — bounded drains, no event loss, honest verdicts, false-reject traps

Browse Source 
Adversarial audit of the PR's own changes (8 finder angles, verified
before fixing). Quartz core:

- fetchAll-family drains get a wall-clock ceiling (maxTotalMs, default
  10x the idle window, delay()-watchdog: cancellable and virtual-time
  testable). The pure idle window was unbounded when a relay trickled
  events forever — sandboxed napplet queries, set -e fetches, and
  marmot await stuck inside one drain. Streaming relays still finish.
- The suspending onEvent hook no longer runs inside a cancellable
  timeout scope (an expiring window could cancel verifyAndStore
  mid-write and silently drop a received event); the timeout is armed
  only when the channels are dry (no per-message timeout-job churn).
- fetchAll is a projection over fetchAllWithHooks: fixes its
  unsynchronized events/seenIds mutation from concurrent socket
  threads and deletes the duplicate loop + per-event activity channel.
- publishAndConfirmDetailed regains its only-responders contract
  (synthetic no-response entries no longer render as 'relay rejected
  your message' in app callers); results built by pure associateWith;
  shared failure-reason constants + PublishResult.isTransportFailure.
- NIP-65 mutations: split read+write r-tags for the same URL now merge
  to BOTH instead of last-wins dropping a facet (+ test).
- TcpProber's 128-thread pool drains after 60s idle.

CLI:

- publishGuard: all-transport failure exits 124 as timeout; rejected/1
  is reserved for an actual OK-false answer.
- --help anywhere in argv is hoisted centrally; 'amy notes post "x"
  --help' prints usage instead of publishing.
- rejectUnknown false-reject traps fixed: geochat --no-fetch behind an
  early return, and 13 elvis-alias short-circuit sites read eagerly.
- Aliases load once per Context and only match name-shaped inputs (no
  shadowing a real npub/NIP-05/hex); stderr color requires a
  positively-known terminal (TERM sniff polluted captured logs).
- Relay-CSV strictness unified on RawEventSupport.relayFlag (post,
  graperank publish/followers/register no longer silently drop
  malformed URLs); Args.timeoutMs(+OrNull) replaces 27 hand-rolled
  conversions, all strict; offer/debit --timeout > 3600 rejected with
  a 'looks like milliseconds' hint; NPub.create idiom; stale jq .id in
  the marmot reactions harness; printUsage drift (offer pay --with,
  profile --clink-offer, search --kind).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CP4kfLCa3wWtE8Khy21Pkj
This commit is contained in:
Claude
2026-07-19 00:17:43 +00:00
parent 962d1706dc
commit 4efb2cca98
43 changed files with 423 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/DEVELOPMENT.md
+3 -2
View File
@@ -294,8 +294,9 @@ contract.
`--json` contract; reuse an existing code before minting a new one:
- **Contract-wide:** `bad_args` (exit 2), `timeout` (exit 124),
`rejected` (every targeted relay refused a publish; payload carries
`event_id` + `rejected_by`), `runtime` (uncaught exception),
`rejected` (every targeted relay refused a publish AND at least one
actually answered `OK false`; payload carries `event_id` + `rejected_by` —
when every failure is transport-level the code is `timeout`/124 instead), `runtime` (uncaught exception),
`invalid_event` (event/template fails id or signature checks),
`http_error`, `bad_response`, `not_found`, `exists`, `read_only`,
`no_identity`, `bad_account`, `signer_error`, `decrypt_failed`.
cli/README.md
+7 -3
View File
@@ -279,7 +279,7 @@ Filter flags are shared by `fetch` and `subscribe`: `--kind K[,K]`, `--author U[
| Command | What it does |
|---|---|
| `amy fetch [filter flags] [--timeout SECS]` | One-shot query — collect until every relay sends EOSE or goes silent for `--timeout` (default 8s — an **idle window**, reset by every arriving event, so a slow-but-streaming relay is never cut off), dedupe, sort newest-first, print and exit. `--limit` defaults to 100. |
| `amy fetch [filter flags] [--timeout SECS]` | One-shot query — collect until every relay sends EOSE or goes silent for `--timeout` (default 8s — an **idle window**, reset by every arriving event, so a slow-but-streaming relay is never cut off; a wall-clock ceiling of 10x the window still bounds a relay that trickles forever), dedupe, sort newest-first, print and exit. `--limit` defaults to 100. |
| `amy fetch CODE [--timeout SECS]` | Code mode — pass a single `nevent`/`naddr`/`nprofile`/`npub`/`note` or `name@domain`. Resolves relays the outbox way: the hints embedded in the code **plus** the author's NIP-65 write relays (draining their kind:10002 on a cache miss), exactly how the app opens a shared link. |
| `amy subscribe [filter flags] [--timeout SECS]` | Live stream — print each matching event as it arrives (NDJSON under `--json`). Runs until `--timeout` SECS or until interrupted. |
| `amy count [filter flags] [--timeout SECS]` | NIP-45 COUNT — per-relay match counts, no event download. |
@@ -732,8 +732,12 @@ Notable error codes (the full canonical list is in
[DEVELOPMENT.md](./DEVELOPMENT.md)):
- **`rejected`** (exit 1) — a publish was refused by **every** targeted
relay; the payload carries `event_id` + `rejected_by`. Partial acceptance
still exits 0 and reports `published_to` / `rejected_by`. `rejected_by`
relay **and at least one relay actually answered `OK false`**; the payload
carries `event_id` + `rejected_by`. When every failure is transport-level
(unreachable, dropped, silent past the window) the code is `timeout`
(exit 124) instead — retry a flaky network, don't give up on a rejection
no relay voiced. Partial acceptance still exits 0 and reports
`published_to` / `rejected_by`. `rejected_by`
is a list of `{relay, reason}` objects — the reason is the relay's own
NIP-01 OK message (`blocked: …`, `rate-limited: …`), a connect error, or
`no response within timeout`, so "why didn't it post?" answers itself:
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Args.kt
+12
View File
@@ -109,6 +109,18 @@ class Args(
?: throw IllegalArgumentException("--$name expects a number, got '$raw'")
}
/** Read `--timeout` (seconds) and return milliseconds; non-numeric input is bad_args. */
fun timeoutMs(defaultSecs: Long): Long = longFlag("timeout", defaultSecs) * 1000
/** Like [timeoutMs] but with no default: null when `--timeout` is absent; non-numeric input is bad_args. */
fun timeoutMsOrNull(): Long? {
val raw = flag("timeout") ?: return null
val secs =
raw.toLongOrNull()
?: throw IllegalArgumentException("--timeout expects a number, got '$raw'")
return secs * 1000
}
fun requireFlag(name: String): String {
consumed.add(name)
return flags[name]
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt
+17 -1
View File
@@ -386,6 +386,22 @@ class Context(
prepared = true
}
/**
* Per-account alias map, read once per invocation. Alias lookups only
* apply to short name-shaped inputs: an npub/nprofile/NIP-05/64-hex input
* always resolves as itself, so an alias can never silently shadow a real
* identifier (`amy dm send bob@damus.io …` reaches the actual NIP-05 owner
* even if a local account happens to carry that name).
*/
private val aliases by lazy { Aliases.load(dataDir) }
private fun aliasFor(input: String): String? {
val nameShaped = input.length in 1..64 && input.all { it.isLetterOrDigit() || it == '_' || it == '-' }
val hexShaped = input.length == 64 && input.all { it.isDigit() || it in 'a'..'f' || it in 'A'..'F' }
if (!nameShaped || hexShaped) return null
return aliases[input]
}
/**
* Resolve an alias (per-account `aliases.json`) / `npub…` / `nprofile…` /
* 64-hex / `name@domain.tld` to a pubkey hex. Aliases are checked first —
@@ -403,7 +419,7 @@ class Context(
*/
suspend fun requireUserHex(input: String): HexKey {
val notResolved = "Could not resolve user: '$input' (accepts an alias, npub, nprofile, 64-hex, or name@domain.tld)"
val resolvable = Aliases.load(dataDir)[input] ?: input
val resolvable = aliasFor(input) ?: input
val hex =
resolveUserHexOrNull(resolvable, nip05Client)
?: throw IllegalArgumentException(notResolved)
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt
+20 -5
View File
@@ -197,7 +197,16 @@ private suspend fun dispatch(argv: Array<String>): Int {
}
val head = filteredArgs[0]
val tail = filteredArgs.drop(1).toTypedArray()
var tail = filteredArgs.drop(1).toTypedArray()
// Central --help hoist: a help request anywhere before a `--` sentinel
// becomes a plain leading --help, so `amy notes post "hi" --help` prints
// usage instead of publishing, and no command can forget to honor it.
// (`--` still lets you pass the literal text: `amy notes post -- --help`.)
val helpIdx = tail.indexOfFirst { it == "--" || it == "--help" || it == "-h" }
if (helpIdx >= 0 && tail[helpIdx] != "--") {
tail = arrayOf("--help")
}
// `use` operates on `<root>/current` directly and must work even
// when account auto-pick would fail (the whole point of `use` is to
@@ -545,7 +554,8 @@ private fun printUsage() {
| [--timeout SECS] (USER: npub|nprofile|hex|name@domain)
|
|Profile (NIP-01 kind:0):
| profile show [USER] [--timeout SECS] fetch latest kind:0 metadata
| profile show [USER] [--refresh] fetch latest kind:0 metadata (--refresh forces
| [--timeout SECS] a relay round-trip past the local cache)
| (USER: npub|nprofile|hex|name@domain)
| profile edit [--name NAME] patch kind:0; unset flags keep prior values,
| [--display-name N] blank values delete the field
@@ -555,6 +565,7 @@ private fun printUsage() {
| [--lud16 X] [--lud06 X]
| [--pronouns P]
| [--twitter H] [--mastodon H] [--github H]
| [--clink-offer NOFFER]
| [--timeout SECS]
|
|Notes (NIP-10 kind:1):
@@ -714,9 +725,13 @@ private fun printUsage() {
|
|CLINK Offers:
| offer info NOFFER decode a noffer1… pointer (local, no network)
| offer discover USER find a user's published offers
| offer pay NOFFER --with NDEBIT request an invoice AND settle it through the
| [--amount SATS] [--timeout SECS] given debit pointer in one step
| offer request NOFFER [--amount SATS] kind:21001 round-trip: ask the service for a
| [--timeout SECS] fresh BOLT11 (amount required for spontaneous
| offers; defaults to the pointer's fixed price)
| [--follow] [--timeout SECS] fresh BOLT11 (amount required for spontaneous
| offers, else the pointer's fixed price;
| --follow waits for settlement)
|
|CLINK Debits:
| debit info NDEBIT decode an ndebit1… pointer (local, no network)
@@ -729,7 +744,7 @@ private fun printUsage() {
| search user QUERY [--limit N] search kind:0 profiles
| [--timeout SECS]
| search note QUERY [--limit N] search event content
| [--kinds K[,K…]] (default kind:1; e.g. 1,30023)
| [--kind K[,K…]] (default kind:1; e.g. 1,30023; --kinds alias)
| [--timeout SECS]
| uses your kind:10007 search-relay
| list, falls back to Amethyst defaults
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Output.kt
+17 -8
View File
@@ -404,14 +404,23 @@ internal class Ansi(
fun forStream(isStderr: Boolean): Ansi {
if (noColor) return Ansi(false)
if (forceColor) return Ansi(true)
// JDK 21 has no per-stream isatty. System.console() != null only when
// BOTH stdin and stdout are terminals, which wrongly stripped color
// from stderr progress during the most common pipe (`amy … | jq`).
// Approximation: stdout keeps the strict console() gate; stderr colors
// whenever an interactive TERM is around (stderr is rarely redirected
// on its own, and NO_COLOR remains the escape hatch).
val tty = System.console() != null || (isStderr && !System.getenv("TERM").isNullOrEmpty())
return Ansi(tty)
// JDK 21 has no per-stream isatty, so the safe rule is: never
// color unless we positively know a terminal is attached.
// System.console() != null covers JDK 21 (both stdin and stdout
// are terminals); on JDK 22+ Console.isTerminal() answers even
// when a stream is redirected (reflective — the method doesn't
// exist on 21). A TERM-sniffing heuristic was tried here and
// reverted: cron/CI capture stderr with TERM exported, and raw
// escape codes in log files are worse than uncolored progress
// during `amy … | jq`. CLICOLOR_FORCE remains the escape hatch.
val console = System.console() ?: return Ansi(false)
val isTerminal =
try {
console.javaClass.getMethod("isTerminal").invoke(console) as? Boolean ?: true
} catch (e: Exception) {
true // JDK 21: console() non-null already implies both-TTY
}
return Ansi(isTerminal)
}
}
}
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BunkerCommand.kt
+2 -2
View File
@@ -172,7 +172,7 @@ object BunkerCommand {
rest: Array<String>,
): Int {
val args = Args(rest)
val timeoutMs = args.flag("timeout")?.toLongOrNull()?.let { it * 1000 }
val timeoutMs = args.timeoutMsOrNull()
interactiveTtyError(args)?.let { return it }
args.rejectUnknown("relay", "secret", "perms")
val accountError = checkHostable(dataDir)
@@ -219,7 +219,7 @@ object BunkerCommand {
rest: Array<String>,
): Int {
val args = Args(rest)
val timeoutMs = args.flag("timeout")?.toLongOrNull()?.let { it * 1000 }
val timeoutMs = args.timeoutMsOrNull()
interactiveTtyError(args)?.let { return it }
args.rejectUnknown("perms")
val uri = args.positional(0, "nostrconnect-uri")
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/ConcordCommands.kt
+4 -2
View File
@@ -96,8 +96,10 @@ object ConcordCommands {
val args = Args(rest)
val name = args.requireFlag("name")
val about = args.flag("about")
// `--relay` is the canonical spelling; `--relays` stays as a silent alias.
val relayArg = parseRelays(args.flag("relay") ?: args.flag("relays"))
// `--relay` is the canonical spelling; `--relays` stays as a silent alias —
// read eagerly so passing both spellings doesn't trip rejectUnknown().
val relaysAlias = args.flag("relays")
val relayArg = parseRelays(args.flag("relay") ?: relaysAlias)
args.rejectUnknown()
Context.open(dataDir).use { ctx ->
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/CountCommand.kt
+1 -1
View File
@@ -58,7 +58,7 @@ object CountCommand {
return 0
}
val args = Args(rest)
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 15L) * 1000
val timeoutMs = args.timeoutMs(15)
val filter = RawEventSupport.buildFilter(args)
args.rejectUnknown("relay")
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/DebitCommands.kt
+6 -2
View File
@@ -99,7 +99,9 @@ object DebitCommands {
val args = Args(rest)
val bolt11 = args.positional(1, "bolt11")
val amount = args.flag("amount")?.toLongOrNull()
val timeoutMs = args.longFlag("timeout", 15) * 1000
val timeoutMs = args.timeoutMs(15)
// Guard ms-era scripts: a value this large is almost certainly milliseconds.
if (timeoutMs > 3_600_000) return Output.error("bad_args", "--timeout is seconds (max 3600); ${timeoutMs / 1000} looks like milliseconds")
args.rejectUnknown()
return roundTrip(dataDir, args, timeoutMs) { client -> client.payInvoice(bolt11, amount) }
@@ -115,7 +117,9 @@ object DebitCommands {
args.flag("amount")?.toLongOrNull()
?: return Output.error("bad_args", "--amount SATS is required for a budget")
val frequency = parseFrequency(args.flag("frequency")) ?: return Output.error("bad_args", "unknown --frequency '${args.flag("frequency")}' (day|week|month)")
val timeoutMs = args.longFlag("timeout", 15) * 1000
val timeoutMs = args.timeoutMs(15)
// Guard ms-era scripts: a value this large is almost certainly milliseconds.
if (timeoutMs > 3_600_000) return Output.error("bad_args", "--timeout is seconds (max 3600); ${timeoutMs / 1000} looks like milliseconds")
args.rejectUnknown()
return roundTrip(dataDir, args, timeoutMs) { client -> client.requestBudget(amount, frequency.value) }
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FetchCommand.kt
+1 -1
View File
@@ -98,7 +98,7 @@ object FetchCommand {
val explicitLimit = args.flag("limit")?.toIntOrNull()
if (explicitLimit != null && explicitLimit < 0) return Output.error("bad_args", "--limit must be >= 0 (0 = unbounded)")
val effectiveLimit: Int? = if (explicitLimit == 0) null else (explicitLimit ?: DEFAULT_LIMIT)
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 8L) * 1000
val timeoutMs = args.timeoutMs(8)
// The filter/relay/paging flags are read later (buildFilter, queryTargets,
// the --paginate branch) and code mode skips them entirely, so whitelist
// them here where both paths still share the flow.
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/FofCommand.kt
+1 -1
View File
@@ -142,7 +142,7 @@ object FofCommand {
val args = Args(rest)
// Overall timeout; per-relay budget is set by OutboxDispatcher's
// default (4s). `--timeout N` overrides the overall cap.
val overallTimeoutMs = args.flag("timeout")?.toLongOrNull()?.times(1000) ?: 8_000L
val overallTimeoutMs = args.timeoutMs(8)
args.rejectUnknown()
Context.open(dataDir).use { ctx ->
ctx.prepare()
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GeochatCommands.kt
+4 -1
View File
@@ -267,12 +267,15 @@ object GeochatCommands {
args: Args,
geohash: String,
): List<NormalizedRelayUrl> {
// Read eagerly so `--relay X --no-fetch` doesn't trip rejectUnknown()
// when the explicit-relay early return skips the directory branch.
val noFetch = args.bool("no-fetch")
val explicit = RawEventSupport.relayFlag(args).toList()
val allExplicit = (explicit + args.positional.mapNotNull { if (it.startsWith("wss://") || it.startsWith("ws://")) RelayUrlNormalizer.normalizeOrNull(it) else null })
if (allExplicit.isNotEmpty()) return allExplicit.distinct()
val directory = GeoRelayDirectory()
if (!args.bool("no-fetch")) {
if (!noFetch) {
runCatching { GeoRelayCsvLoader { OkHttpClient() }.refresh(directory) }
}
return directory.closestRelays(geohash)
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GitCommands.kt
+4 -2
View File
@@ -87,8 +87,10 @@ object GitCommands {
): Int {
val args = Args(rest)
val name = args.flag("name") ?: return Output.error("bad_args", "git announce requires --name")
// `--identifier` is the spelled-out alias of `--d` (the d-tag).
val identifier = args.flag("d") ?: args.flag("identifier") ?: name
// `--identifier` is the spelled-out alias of `--d` (the d-tag) — read
// eagerly so passing both spellings doesn't trip rejectUnknown().
val identifierAlias = args.flag("identifier")
val identifier = args.flag("d") ?: identifierAlias ?: name
val csv = { key: String ->
args
.flag(key)
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/KeyCommands.kt
+6 -2
View File
@@ -110,7 +110,9 @@ object KeyCommands {
private fun encrypt(rest: Array<String>): Int {
val args = Args(rest)
val priv = privHexOrNull(args.positional(0, "secret-key").trim()) ?: return Output.error("bad_args", "expected an nsec or 64-char hex secret key")
val password = args.flag("password") ?: args.flag("pw") ?: return Output.error("bad_args", "key encrypt requires --password")
// Read both spellings eagerly so passing both doesn't trip rejectUnknown().
val pwAlias = args.flag("pw")
val password = args.flag("password") ?: pwAlias ?: return Output.error("bad_args", "key encrypt requires --password")
args.rejectUnknown()
val ncryptsec = Nip49().encrypt(priv, password)
Output.emit(mapOf("ncryptsec" to ncryptsec))
@@ -121,7 +123,9 @@ object KeyCommands {
val args = Args(rest)
val ncryptsec = args.positional(0, "ncryptsec").trim()
if (!ncryptsec.startsWith("ncryptsec")) return Output.error("bad_args", "expected an ncryptsec1… string")
val password = args.flag("password") ?: args.flag("pw") ?: return Output.error("bad_args", "key decrypt requires --password")
// Read both spellings eagerly so passing both doesn't trip rejectUnknown().
val pwAlias = args.flag("pw")
val password = args.flag("password") ?: pwAlias ?: return Output.error("bad_args", "key decrypt requires --password")
args.rejectUnknown()
val privHex =
try {
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/LoginCommand.kt
+3 -1
View File
@@ -152,8 +152,10 @@ object LoginCommand {
if (key.startsWith("bunker://")) return Identity.fromBunkerUri(key)
// 1. ncryptsec — password mandatory.
if (key.startsWith("ncryptsec")) {
// Read both spellings eagerly so passing both doesn't trip rejectUnknown().
val pwAlias = args.flag("pw")
val pw =
args.flag("password") ?: args.flag("pw")
args.flag("password") ?: pwAlias
?: throw IllegalArgumentException("ncryptsec input requires --password")
val privHex = Nip49().decrypt(key, pw)
return Identity.fromPrivateKey(
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/NappletCommands.kt
+4 -2
View File
@@ -152,7 +152,8 @@ object NappletCommands {
): Int {
val args = Args(rest)
val author = args.positionalOrNull(0) ?: return Output.error("bad_args", "napplet serve <author> [--d ID] [--port N] [--server S] [--relay R]")
val identifier = args.flag("d") ?: args.flag("identifier")
val identifierAlias = args.flag("identifier")
val identifier = args.flag("d") ?: identifierAlias
val port = args.intFlag("port", 8080)
val extraServers = StaticSiteFetch.commaList(args.flag("server"))
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
@@ -210,7 +211,8 @@ object NappletCommands {
if (snapshotId == null && author == null) {
return Output.error("bad_args", "napplet fetch <author> [--d ID] | --snapshot <event-id> [--path P]")
}
val identifier = args.flag("d") ?: args.flag("identifier")
val identifierAlias = args.flag("identifier")
val identifier = args.flag("d") ?: identifierAlias
val requestPath = args.flag("path", "/")!!
val outFile = args.flag("out")
val timeoutSecs = args.longFlag("timeout", 8L)
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/NipCommand.kt
+1 -1
View File
@@ -82,7 +82,7 @@ object NipCommand {
val args = Args(rest)
val raw = args.positional(0, "nip-number").trim()
val slug = normalizeSlug(raw)
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 8L) * 1000
val timeoutMs = args.timeoutMs(8)
args.rejectUnknown()
// 1. Canonical repo first.
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/NostrConnect.kt
+1 -1
View File
@@ -92,7 +92,7 @@ object NostrConnect {
}
val relays = RawEventSupport.relayFlag(args).ifEmpty { DefaultNIP65RelaySet }
if (relays.isEmpty()) return Output.error("bad_args", "no relays; pass --relay URL[,URL…]")
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 120L) * 1000
val timeoutMs = args.timeoutMs(120)
val name = args.flag("name")
// Optional NIP-46 permission request (`--perms sign_event:1,nip44_encrypt,…`). When present it
// rides along in the offer so a signer that honors `perms` (e.g. Amethyst's informed-consent
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/NsiteCommands.kt
+4 -2
View File
@@ -162,7 +162,8 @@ object NsiteCommands {
): Int {
val args = Args(rest)
val author = args.positionalOrNull(0) ?: return Output.error("bad_args", "nsite serve <author> [--d ID] [--port N] [--server S] [--relay R]")
val identifier = args.flag("d") ?: args.flag("identifier")
val identifierAlias = args.flag("identifier")
val identifier = args.flag("d") ?: identifierAlias
val port = args.intFlag("port", 8080)
val extraServers = StaticSiteFetch.commaList(args.flag("server"))
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
@@ -219,7 +220,8 @@ object NsiteCommands {
): Int {
val args = Args(rest)
val author = args.positionalOrNull(0) ?: return Output.error("bad_args", "nsite fetch <author> [--d ID] [--path P]")
val identifier = args.flag("d") ?: args.flag("identifier")
val identifierAlias = args.flag("identifier")
val identifier = args.flag("d") ?: identifierAlias
val requestPath = args.flag("path", "/")!!
val outFile = args.flag("out")
val timeoutSecs = args.longFlag("timeout", 8L)
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/OfferCommands.kt
+6 -2
View File
@@ -152,7 +152,9 @@ object OfferCommands {
): Int {
val args = Args(rest)
val amount = args.flag("amount")?.toLongOrNull()
val timeoutMs = args.longFlag("timeout", 15) * 1000
val timeoutMs = args.timeoutMs(15)
// Guard ms-era scripts: a value this large is almost certainly milliseconds.
if (timeoutMs > 3_600_000) return Output.error("bad_args", "--timeout is seconds (max 3600); ${timeoutMs / 1000} looks like milliseconds")
val follow = args.bool("follow")
// Offers can be configured to require payer fields (e.g. email); Lightning.Pub
// rejects a request missing them as "Invalid Offer" (code 1), so the round-trip
@@ -230,7 +232,9 @@ object OfferCommands {
): Int {
val args = Args(rest)
val amount = args.flag("amount")?.toLongOrNull()
val timeoutMs = args.longFlag("timeout", 15) * 1000
val timeoutMs = args.timeoutMs(15)
// Guard ms-era scripts: a value this large is almost certainly milliseconds.
if (timeoutMs > 3_600_000) return Output.error("bad_args", "--timeout is seconds (max 3600); ${timeoutMs / 1000} looks like milliseconds")
val offer =
ClinkPointerParser.parse(args.positional(0, "noffer").trim()) as? NOffer
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/OutboxCommand.kt
+4 -5
View File
@@ -24,9 +24,8 @@ import com.vitorpamplona.amethyst.cli.Args
import com.vitorpamplona.amethyst.cli.Context
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip19Bech32.toNpub
import com.vitorpamplona.quartz.nip19Bech32.entities.NPub
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
/**
@@ -58,7 +57,7 @@ object OutboxCommand {
val args = Args(rest)
val user = args.positional(0, "user")
val refresh = args.bool("refresh")
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 8L) * 1000
val timeoutMs = args.timeoutMs(8)
args.rejectUnknown()
Context.openOrAnonymous(dataDir).use { ctx ->
@@ -79,14 +78,14 @@ object OutboxCommand {
}
if (event == null) {
Output.emit(mapOf("pubkey" to pubkey, "npub" to pubkey.hexToByteArray().toNpub(), "found" to false))
Output.emit(mapOf("pubkey" to pubkey, "npub" to NPub.create(pubkey), "found" to false))
return 0
}
Output.emit(
mapOf(
"pubkey" to pubkey,
"npub" to pubkey.hexToByteArray().toNpub(),
"npub" to NPub.create(pubkey),
"found" to true,
"created_at" to event.createdAt,
"read" to (event.readRelaysNorm()?.map { it.url } ?: emptyList()),
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/Podcast20Commands.kt
+4 -2
View File
@@ -161,7 +161,8 @@ object Podcast20Commands {
return Output.error("bad_args", "podcast20 episode --value-json is not valid JSON")
}
val dTag = args.flag("d") ?: args.flag("identifier") ?: generateDTag("episode")
val identifierAlias = args.flag("identifier")
val dTag = args.flag("d") ?: identifierAlias ?: generateDTag("episode")
val video = args.flag("video")?.let { PodcastAudio(it, args.flag("video-type")) }
Context.open(dataDir).use { ctx ->
@@ -208,7 +209,8 @@ object Podcast20Commands {
val args = Args(rest)
val title = args.flag("title") ?: return Output.error("bad_args", "podcast20 trailer requires --title")
val url = args.flag("url") ?: return Output.error("bad_args", "podcast20 trailer requires --url")
val dTag = args.flag("d") ?: args.flag("identifier") ?: generateDTag("trailer")
val identifierAlias = args.flag("identifier")
val dTag = args.flag("d") ?: identifierAlias ?: generateDTag("trailer")
Context.open(dataDir).use { ctx ->
ctx.prepare()
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/PostCommand.kt
+4 -9
View File
@@ -24,7 +24,6 @@ import com.vitorpamplona.amethyst.cli.Args
import com.vitorpamplona.amethyst.cli.Context
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
import com.vitorpamplona.quartz.nip13Pow.miner.PoWMiner
import com.vitorpamplona.quartz.nip13Pow.pow
@@ -58,12 +57,9 @@ object PostCommand {
?: return Output.error("bad_args", "post <text> [--relay URL …] [--pow BITS [--pow-timeout SECS]]")
if (text.isBlank()) return Output.error("bad_args", "post text must not be blank")
val extraRelays =
args
.flag("relay")
?.split(',')
?.map { it.trim() }
?.filter { it.isNotEmpty() } ?: emptyList()
// Strictly validated like every other `--relay` in amy — a malformed
// entry is a bad_args failure, not a silent drop.
val extraRelays = RawEventSupport.relayFlag(args)
val powRaw = args.flag("pow")
val powTarget = powRaw?.toIntOrNull()
@@ -76,8 +72,7 @@ object PostCommand {
Context.open(dataDir).use { ctx ->
ctx.prepare()
val outbox = ctx.outboxRelays()
val extraNormalized = extraRelays.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }
val targets = (outbox + extraNormalized).toSet()
val targets = (outbox + extraRelays).toSet()
if (targets.isEmpty()) {
return Output.error("no_relays", "no outbox relays configured; pass --relay or run `amy relay add`")
}
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/ProfileCommands.kt
+3 -4
View File
@@ -27,11 +27,10 @@ import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.quartz.experimental.clink.pointers.ClinkPointerParser
import com.vitorpamplona.quartz.experimental.clink.pointers.NOffer
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip19Bech32.toNpub
import com.vitorpamplona.quartz.nip19Bech32.entities.NPub
/**
* `amy profile <show|edit>` — read and update the current user's
@@ -118,7 +117,7 @@ object ProfileCommands {
Output.emit(
mapOf(
"pubkey" to pubKey,
"npub" to pubKey.hexToByteArray().toNpub(),
"npub" to NPub.create(pubKey),
"found" to false,
"source" to source,
"queried_relays" to queried.map { it.url },
@@ -135,7 +134,7 @@ object ProfileCommands {
Output.emit(
mapOf(
"pubkey" to pubKey,
"npub" to pubKey.hexToByteArray().toNpub(),
"npub" to NPub.create(pubKey),
"found" to true,
"source" to source,
"event_id" to event.id,
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RawEventSupport.kt
+18 -5
View File
@@ -77,11 +77,24 @@ object RawEventSupport {
eventId: String,
): Int? {
if (ack.isEmpty() || ack.any { it.value.accepted }) return null
return Output.error(
"rejected",
"no relay accepted event $eventId",
extra = mapOf("event_id" to eventId, "rejected_by" to rejectedBy(ack)),
)
// `rejected` means a relay actually answered `OK false`. When every
// failure is transport-level (silent, unreachable, dropped) the honest
// code is `timeout` (exit 124) — a retry-on-124 script should retry a
// flaky network, not give up on a "rejection" no relay ever voiced.
val genuinelyRejected = ack.values.any { !it.isTransportFailure }
return if (genuinelyRejected) {
Output.error(
"rejected",
"no relay accepted event $eventId",
extra = mapOf("event_id" to eventId, "rejected_by" to rejectedBy(ack)),
)
} else {
Output.error(
"timeout",
"no relay answered for event $eventId (all unreachable or silent)",
extra = mapOf("event_id" to eventId, "rejected_by" to rejectedBy(ack)),
)
}
}
/**
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt
+1 -1
View File
@@ -333,7 +333,7 @@ object RelayCommands {
val args = Args(rest)
// Per probe WAVE, not per relay or total — a wave's stragglers are cut off
// together when it elapses.
val timeoutMs = args.longFlag("timeout", 15L) * 1000
val timeoutMs = args.timeoutMs(15)
// Relays dialed at once; --relay-concurrency accepted as the alias the
// graperank verbs spell it with.
val waveSize = args.intFlag("concurrency", args.intFlag("relay-concurrency", Context.defaultPreconnectCap))
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SearchCommand.kt
+6 -4
View File
@@ -86,7 +86,7 @@ object SearchCommand {
val query = rest[0]
val args = Args(rest.drop(1).toTypedArray())
val limit = args.longFlag("limit", 50L).toInt()
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8)
args.rejectUnknown()
val filter =
@@ -127,10 +127,12 @@ object SearchCommand {
val query = rest[0]
val args = Args(rest.drop(1).toTypedArray())
val limit = args.longFlag("limit", 50L).toInt()
val timeoutMs = args.longFlag("timeout", 8L) * 1000
// `--kind` is canonical (matching fetch/subscribe); `--kinds` stays a silent alias.
val timeoutMs = args.timeoutMs(8L)
// `--kind` is canonical (matching fetch/subscribe); `--kinds` stays a silent
// alias — read eagerly so passing both spellings doesn't trip rejectUnknown().
val kindsAlias = args.flag("kinds")
val kindList =
(args.flag("kind") ?: args.flag("kinds"))
(args.flag("kind") ?: kindsAlias)
?.split(',')
?.mapNotNull { it.trim().toIntOrNull() }
?.takeIf { it.isNotEmpty() }
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/StaticSitePublish.kt
+2 -1
View File
@@ -66,7 +66,8 @@ object StaticSitePublish {
val servers = StaticSiteFetch.commaList(args.flag("server"))
if (servers.isEmpty()) return Output.error("bad_args", "publish requires --server <blossom-url> (comma-separated for mirrors)")
val identifier = args.flag("d") ?: args.flag("identifier")
val identifierAlias = args.flag("identifier")
val identifier = args.flag("d") ?: identifierAlias
val requires = StaticSiteFetch.commaList(args.flag("requires"))
val title = args.flag("title")
val description = args.flag("description")
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SubscribeCommand.kt
+1 -1
View File
@@ -69,7 +69,7 @@ object SubscribeCommand {
return 0
}
val args = Args(rest)
val timeoutMs = args.flag("timeout")?.toLongOrNull()?.let { it * 1000 }
val timeoutMs = args.timeoutMsOrNull()
val filter = RawEventSupport.buildFilter(args)
args.rejectUnknown("relay")
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt
+1 -1
View File
@@ -133,7 +133,7 @@ object SyncCommand {
val relay =
RelayUrlNormalizer.normalizeOrNull(relayUrl)
?: return Output.invalidRelayUrl(relayUrl)
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 30L) * 1000
val timeoutMs = args.timeoutMs(30)
// Default direction is download; --up adds upload.
val up = args.bool("up")
val down = args.bool("down") || !up
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/ZapCommand.kt
+2 -2
View File
@@ -103,7 +103,7 @@ object ZapCommand {
val args = Args(rest.drop(2).toTypedArray())
val comment = args.flag("comment") ?: ""
val zapType = parseZapType(args)
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8)
val withFlag = args.flag("with")
val settleWith =
if (withFlag == null) {
@@ -152,7 +152,7 @@ object ZapCommand {
val args = Args(rest.drop(2).toTypedArray())
val comment = args.flag("comment") ?: ""
val zapType = parseZapType(args)
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8)
val withFlag = args.flag("with")
val settleWith =
if (withFlag == null) {
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankCrawl.kt
+7 -9
View File
@@ -24,6 +24,7 @@ import com.vitorpamplona.amethyst.cli.Args
import com.vitorpamplona.amethyst.cli.Context
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.amethyst.cli.commands.RawEventSupport
import com.vitorpamplona.amethyst.commons.defaults.Constants
import com.vitorpamplona.amethyst.commons.defaults.DefaultIndexerRelayList
import com.vitorpamplona.quartz.experimental.graperank.FollowerCrawler
@@ -32,7 +33,6 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent
@@ -97,7 +97,7 @@ object GrapeRankCrawl {
contentAggregatorRelays = aggregators,
maxRounds = args.intFlag("max-rounds", Int.MAX_VALUE),
maxHops = args.intFlag("max-hops", Int.MAX_VALUE),
timeoutMs = args.longFlag("timeout", 10L) * 1000,
timeoutMs = args.timeoutMs(10),
parkTimeoutMs = args.longFlag("park-timeout", 40L) * 1000,
diagnose = args.bool("diagnose"),
insertBatchSize = args.intFlag(FLAG_INSERT_BATCH, INSERT_BATCH_DEFAULT),
@@ -251,7 +251,9 @@ object GrapeRankCrawl {
): Int {
val args = Args(rest)
val observerArg = args.positionalOrNull(0)
val relayArg = args.flag("relay")
// Strictly validated like every other `--relay` in amy — a malformed
// entry is a bad_args failure, not a silent drop.
val explicitRelays = RawEventSupport.relayFlag(args)
val relayConcurrency = args.intFlag(FLAG_RELAY_CONCURRENCY, args.intFlag(FLAG_CONCURRENCY, 16))
// --max/--timeout/--insert-batch are read later inside the crawler
// Config; --no-reachability-cache inside allKnownRelays.
@@ -267,11 +269,7 @@ object GrapeRankCrawl {
val observer = observerArg?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
val relays =
relayArg
?.split(",")
?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it.trim()) }
?.toSet()
?.takeIf { it.isNotEmpty() }
explicitRelays.takeIf { it.isNotEmpty() }
?: allKnownRelays(ctx, args)
if (relays.isEmpty()) {
@@ -289,7 +287,7 @@ object GrapeRankCrawl {
// Default null → pull EVERY follower each relay holds; --max
// N caps the total per relay for a quick spot check.
maxPerRelay = args.flag("max")?.toIntOrNull(),
timeoutMs = args.longFlag("timeout", 15L) * 1000,
timeoutMs = args.timeoutMs(15),
maxConcurrentRelays = relayConcurrency,
insertBatchSize = args.intFlag(FLAG_INSERT_BATCH, INSERT_BATCH_DEFAULT),
),
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankOperator.kt
+13 -7
View File
@@ -191,7 +191,7 @@ object GrapeRankOperator {
relayConcurrency = args.intFlag(FLAG_RELAY_CONCURRENCY, args.intFlag(FLAG_CONCURRENCY, 4)),
authorChunk = args.intFlag("author-chunk", 500),
minAuthors = args.intFlag("min-authors", 1),
idleTimeoutMs = args.longFlag("timeout", 30L) * 1000,
idleTimeoutMs = args.timeoutMs(30),
knownDead = knownDead,
),
log = { System.err.println(it) },
@@ -325,11 +325,14 @@ object GrapeRankOperator {
rest: Array<String>,
): Int {
val args = Args(rest)
val providerArg = args.positionalOrNull(0) ?: args.flag("provider")
// Read `--provider` eagerly so a positional PROVIDER plus the flag
// spelling doesn't trip rejectUnknown().
val providerFlag = args.flag("provider")
val providerArg = args.positionalOrNull(0) ?: providerFlag
val serviceArg = args.flag("service")
val relayArg = args.flag("relay")
val isPrivate = args.bool("private")
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8L)
args.rejectUnknown()
val service =
@@ -344,7 +347,7 @@ object GrapeRankOperator {
val outbox = ctx.outboxRelays()
val relay =
relayArg?.let { RelayUrlNormalizer.normalizeOrNull(it) }
relayArg?.let { RelayUrlNormalizer.normalizeOrNull(it) ?: throw IllegalArgumentException("invalid relay url: $it") }
?: outbox.firstOrNull()
?: return Output.error("no_relays", "no relay hint; pass --relay URL or configure outbox relays")
@@ -407,13 +410,16 @@ object GrapeRankOperator {
rest: Array<String>,
): Int {
val args = Args(rest)
// Read `--provider` eagerly so a positional PROVIDER plus the flag
// spelling doesn't trip rejectUnknown().
val providerFlag = args.flag("provider")
val providerArg =
args.positionalOrNull(0)
?: args.flag("provider")
?: providerFlag
?: return Output.error("bad_args", "usage: amy graperank unregister PROVIDER [--service KIND:TAG] [--relay URL]")
val serviceArg = args.flag("service")
val relayArg = args.flag("relay")
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8L)
args.rejectUnknown()
val service =
@@ -497,7 +503,7 @@ object GrapeRankOperator {
val args = Args(rest)
val userArg = args.positionalOrNull(0)
val refresh = args.bool("refresh")
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8)
args.rejectUnknown()
Context.open(dataDir).use { ctx ->
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/graperank/GrapeRankPublish.kt
+7 -9
View File
@@ -24,6 +24,7 @@ import com.vitorpamplona.amethyst.cli.Args
import com.vitorpamplona.amethyst.cli.Context
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.amethyst.cli.commands.RawEventSupport
import com.vitorpamplona.amethyst.commons.defaults.Constants
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankPublisher
import com.vitorpamplona.quartz.nip01Core.core.Event
@@ -31,7 +32,6 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.TrustProviderListEvent
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.serviceProviders
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ProviderTypes
@@ -61,12 +61,14 @@ object GrapeRankPublish {
): Int {
val args = Args(rest)
val observerArg = args.positionalOrNull(0)
val relayArg = args.flag("relay")
// Strictly validated like every other `--relay` in amy — a malformed
// entry is a bad_args failure, not a silent drop.
val explicitRelays = RawEventSupport.relayFlag(args)
// --relay-concurrency is canonical for "relays worked at once" across the
// graperank verbs; --concurrency is accepted everywhere as its alias.
val relayConcurrency = args.intFlag(FLAG_RELAY_CONCURRENCY, args.intFlag(FLAG_CONCURRENCY, 4))
// Idle watchdog per relay reconcile (not a total budget), like `refresh`.
val idleTimeoutMs = args.longFlag("timeout", 30L) * 1000
val idleTimeoutMs = args.timeoutMs(30)
args.rejectUnknown()
Context.open(dataDir).use { ctx ->
@@ -77,11 +79,7 @@ object GrapeRankPublish {
// Cards live on the operator's own relay(s); --relay overrides.
val relays =
relayArg
?.split(",")
?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it.trim()) }
?.toSet()
?.takeIf { it.isNotEmpty() }
explicitRelays.takeIf { it.isNotEmpty() }
?: opKeys.operatorRelays()
if (relays.isEmpty()) {
return Output.error("no_relays", "no operator relay configured — run `amy graperank operator relay <url>` or pass --relay")
@@ -162,7 +160,7 @@ object GrapeRankPublish {
?: return Output.error("bad_args", "usage: amy graperank rank USER [--provider PUBKEY] [--refresh] [--timeout SECS]")
val providerArg = args.flag("provider")
val refresh = args.bool("refresh")
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val timeoutMs = args.timeoutMs(8)
args.rejectUnknown()
Context.openOrAnonymous(dataDir).use { ctx ->
cli/tests/marmot/tests-extras.sh
+1 -1
View File
@@ -47,7 +47,7 @@ test_09_reply_react_unreact() {
sleep 3
local a_anchor_id
a_anchor_id=$(amy_json marmot message list "$gid" --limit 50 2>/dev/null \
| jq -r '[.messages[]? | select((.content // "") == "anchor for reactions")][0].id // empty')
| jq -r '[.messages[]? | select((.content // "") == "anchor for reactions")][0].event_id // empty')
if [[ -z "$a_anchor_id" || "$a_anchor_id" == "null" ]]; then
record_result "$id" fail "amy couldn't find anchor message in local log"; return
fi
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllExt.kt
+16 -79
View File
@@ -23,14 +23,10 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.withTimeoutOrNull
suspend fun INostrClient.fetchAll(
relay: String,
@@ -78,88 +74,29 @@ suspend fun INostrClient.fetchAll(
* [timeoutMs] is an **idle window, not a hard cap**: every arriving event or
* terminal signal resets it, so a slow relay actively streaming a large
* backlog is never cropped mid-delivery. The fetch only gives up after a full
* window of silence — the timeout's job is detecting relays that will never
* reach a terminal state, not bounding total work.
* window of silence — or at the [maxTotalMs] wall-clock ceiling (default 10x
* the idle window), which keeps a trickling never-terminal relay from pinning
* the caller forever.
*
* Thin projection over [fetchAllWithHooks] — one shared loop implementation,
* with dedup done in the (single-threaded) hook so no shared collection is
* ever touched from socket callback threads.
*/
suspend fun INostrClient.fetchAll(
subscriptionId: String = newSubId(),
filters: Map<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 30_000L,
maxTotalMs: Long = timeoutMs * 10,
): List<Event> {
val doneChannel = Channel<NormalizedRelayUrl>(Channel.UNLIMITED)
// Conflated liveness ping: onEvent runs on the socket thread and only needs
// to signal "still streaming" — one pending token is enough to reset the
// idle window, so repeats are safely dropped.
val activityChannel = Channel<Unit>(Channel.CONFLATED)
val events = mutableListOf<Event>()
val seenIds = mutableSetOf<HexKey>()
val remaining = filters.keys.toMutableSet()
val listener =
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
if (seenIds.add(event.id)) {
events.add(event)
}
activityChannel.trySend(Unit)
}
override fun onCannotConnect(
relay: NormalizedRelayUrl,
message: String,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
}
override fun onClosed(
message: String,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
}
}
try {
subscribe(subscriptionId, filters, listener)
// Idle-window wait: each pass waits up to [timeoutMs] for the next
// signal — a terminal message or an event-activity ping. Progress of
// either kind restarts the window; a full window of silence ends the
// fetch with whatever arrived.
while (remaining.isNotEmpty()) {
val progressed =
withTimeoutOrNull(timeoutMs) {
select<Unit> {
doneChannel.onReceive { finished -> remaining.remove(finished) }
activityChannel.onReceive { }
}
}
if (progressed == null) break
}
} finally {
unsubscribe(subscriptionId)
doneChannel.close()
activityChannel.close()
}
return events.sortedWith(DefaultFeedOrderEvent)
return fetchAllWithHooks(
filters = filters,
timeoutMs = timeoutMs,
subscriptionId = subscriptionId,
maxTotalMs = maxTotalMs,
) { _, event -> seenIds.add(event.id) }
.map { it.second }
.sortedWith(DefaultFeedOrderEvent)
}
val DefaultFeedOrderEvent: Comparator<Event> =
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllWithHooksExt.kt
+97 -31
View File
@@ -31,6 +31,7 @@ import com.vitorpamplona.quartz.utils.SeenIds
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.withTimeoutOrNull
@@ -48,7 +49,9 @@ import kotlinx.coroutines.withTimeoutOrNull
* cropped mid-delivery — the fetch ends when the work is done or when nothing
* has arrived for [timeoutMs] (a stall). The terminal conditions (EOSE /
* CLOSED / cannot-connect per relay) are what bound the fetch; the timeout's
* only job is detecting relays that will never reach one.
* only job is detecting relays that will never reach one. [maxTotalMs]
* (default 10x the idle window) is the wall-clock ceiling that keeps a
* trickling relay from pinning the caller forever.
*
* Extras over [fetchAll]:
* - **[onEvent] hook** — suspending per-event callback, invoked single-threaded
@@ -77,6 +80,15 @@ suspend fun INostrClient.fetchAllWithHooks(
pendingOnAuthRequired: Boolean = false,
deadOut: MutableMap<NormalizedRelayUrl, DrainFailure>? = null,
onTimeout: ((stalled: Set<NormalizedRelayUrl>, doneReasons: Map<NormalizedRelayUrl, String>, collected: List<Pair<NormalizedRelayUrl, Event>>) -> Unit)? = null,
/**
* Hard wall-clock ceiling. The idle window alone is unbounded when a relay
* keeps trickling events without ever reaching a terminal state — an
* adversarial or misbehaving relay could pin the caller forever. The cap
* restores an upper bound while staying far above the idle window, so a
* legitimately streaming relay still finishes its backlog. Pass
* [Long.MAX_VALUE] for a deliberately uncapped drain.
*/
maxTotalMs: Long = timeoutMs * 10,
onEvent: suspend (relay: NormalizedRelayUrl, event: Event) -> Boolean,
): List<Pair<NormalizedRelayUrl, Event>> {
if (filters.isEmpty()) return emptyList()
@@ -124,40 +136,94 @@ suspend fun INostrClient.fetchAllWithHooks(
}
}
val collected = mutableListOf<Pair<NormalizedRelayUrl, Event>>()
// One conflated token, armed by a delay()-based watchdog, ends the fetch
// at the wall-clock ceiling. delay() keeps the cap on the coroutine clock
// (cancellable, virtual-time-testable) instead of sampling a wall clock.
val capChannel = Channel<Unit>(Channel.CONFLATED)
try {
subscribe(subscriptionId, filters, listener)
// Idle-window wait: each pass waits up to [timeoutMs] for the NEXT
// message; any event or terminal signal restarts the window. Only a
// full window of silence ends the fetch early.
var stalled = false
while (remaining.isNotEmpty()) {
val progressed =
withTimeoutOrNull(timeoutMs) {
select<Unit> {
eventChannel.onReceive { pair ->
if (onEvent(pair.first, pair.second)) collected.add(pair)
}
doneChannel.onReceive { (relay, reason) ->
remaining.remove(relay)
doneReasons[relay] = reason
}
coroutineScope {
subscribe(subscriptionId, filters, listener)
val watchdog =
if (maxTotalMs == Long.MAX_VALUE) {
null
} else {
launch {
delay(maxTotalMs)
capChannel.trySend(Unit)
}
}
if (progressed == null) {
stalled = true
break
// Idle-window wait with a wall-clock ceiling. Two structural rules:
//
// 1. The suspending [onEvent] hook NEVER runs inside a timeout
// scope. Cancellation only lands at suspension points, so a
// hook stalled in verify/persist work would otherwise be
// cancelled mid-write by an expiring window and the
// already-received event silently lost. The select bodies
// below only stash/bookkeep (non-suspending — they cannot be
// cancelled mid-body); the hook runs after.
//
// 2. The timeout is only armed when both channels are DRY.
// Buffered messages drain through the tryReceive fast path
// with zero timeout-job churn — under burst arrival a
// per-message withTimeoutOrNull would pay one
// scheduled+cancelled cancellation task per event for nothing.
var stalled = false
var capped = false
while (remaining.isNotEmpty() && !capped) {
var pending: Pair<NormalizedRelayUrl, Event>? = null
// Fast path: consume whatever is already buffered.
if (capChannel.tryReceive().isSuccess) {
capped = true
break
}
val bufferedDone = doneChannel.tryReceive().getOrNull()
if (bufferedDone != null) {
remaining.remove(bufferedDone.first)
doneReasons[bufferedDone.first] = bufferedDone.second
continue
}
pending = eventChannel.tryReceive().getOrNull()
// Slow path: both dry — arm one idle wait for the next signal.
if (pending == null) {
val progressed =
withTimeoutOrNull(timeoutMs) {
select<Unit> {
eventChannel.onReceive { pending = it }
doneChannel.onReceive { (relay, reason) ->
remaining.remove(relay)
doneReasons[relay] = reason
}
capChannel.onReceive { capped = true }
}
}
if (progressed == null) {
stalled = true
break
}
if (capped) break
}
pending?.let { pair ->
if (onEvent(pair.first, pair.second)) collected.add(pair)
}
}
}
// Drain any events that landed after the last terminal signal (or
// during the final window) but before unsubscribe.
while (true) {
val r = eventChannel.tryReceive()
if (!r.isSuccess) break
val pair = r.getOrThrow()
if (onEvent(pair.first, pair.second)) collected.add(pair)
}
if (stalled && remaining.isNotEmpty()) {
onTimeout?.invoke(remaining, doneReasons, collected)
// Drain any events that landed after the last terminal signal (or
// during the final window) but before unsubscribe. Skipped when
// the ceiling fired — the cap must actually stop the work.
if (!capped) {
while (true) {
val r = eventChannel.tryReceive()
if (!r.isSuccess) break
val pair = r.getOrThrow()
if (onEvent(pair.first, pair.second)) collected.add(pair)
}
}
if ((stalled || capped) && remaining.isNotEmpty()) {
onTimeout?.invoke(remaining, doneReasons, collected)
}
watchdog?.cancel()
}
} finally {
unsubscribe(subscriptionId)
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt
+31 -11
View File
@@ -45,7 +45,22 @@ import kotlinx.coroutines.withTimeoutOrNull
class PublishResult(
val accepted: Boolean,
val message: String,
)
) {
/**
* True when this failure came from the transport (never connected,
* dropped, or silent past the timeout) rather than from the relay
* actually answering `OK false`. Callers use this to tell "the relay
* refused the event" apart from "the relay never weighed in".
*/
val isTransportFailure: Boolean
get() = !accepted && (message == NO_RESPONSE || message == DISCONNECTED || message.startsWith(CANNOT_CONNECT_PREFIX))
companion object {
const val NO_RESPONSE = "no response within timeout"
const val DISCONNECTED = "disconnected before OK"
const val CANNOT_CONNECT_PREFIX = "cannot connect: "
}
}
@OptIn(DelicateCoroutinesApi::class)
suspend fun INostrClient.publishAndConfirm(
@@ -57,15 +72,22 @@ suspend fun INostrClient.publishAndConfirm(
/**
* Sends an event to the given relays and waits for OK responses.
* Returns per-relay results: relay URL -> accepted (true/false).
* Prefer [publishAndCollectResults] when the caller can surface the
* relays' rejection reasons — this projection drops them.
* Keeps the historical contract: only relays that RESPONDED (an OK, a
* connect failure, or a disconnect) appear — a relay that stayed silent
* past the timeout is absent, not reported as `false`, so long-standing
* callers that render the false entries as "rejected by" don't start
* blaming relays that merely never answered. Prefer
* [publishAndCollectResults] when the caller can surface the reasons.
*/
@OptIn(DelicateCoroutinesApi::class)
suspend fun INostrClient.publishAndConfirmDetailed(
event: Event,
relayList: Set<NormalizedRelayUrl>,
timeoutInSeconds: Long = 15,
): Map<NormalizedRelayUrl, Boolean> = publishAndCollectResults(event, relayList, timeoutInSeconds).mapValues { it.value.accepted }
): Map<NormalizedRelayUrl, Boolean> =
publishAndCollectResults(event, relayList, timeoutInSeconds)
.filterValues { it.message != PublishResult.NO_RESPONSE }
.mapValues { it.value.accepted }
/**
* Sends an event to the given relays and waits for OK responses, keeping the
@@ -90,14 +112,14 @@ suspend fun INostrClient.publishAndCollectResults(
errorMessage: String,
) {
if (relay.url in relayList) {
resultChannel.trySend(DetailedResult(relay.url, false, "cannot connect: $errorMessage"))
resultChannel.trySend(DetailedResult(relay.url, false, PublishResult.CANNOT_CONNECT_PREFIX + errorMessage))
Log.d("publishAndConfirm") { "Error from relay ${relay.url}: $errorMessage" }
}
}
override fun onDisconnected(relay: IRelayClient) {
if (relay.url in relayList) {
resultChannel.trySend(DetailedResult(relay.url, false, "disconnected before OK"))
resultChannel.trySend(DetailedResult(relay.url, false, PublishResult.DISCONNECTED))
Log.d("publishAndConfirm") { "Disconnected from relay ${relay.url}" }
}
}
@@ -160,11 +182,9 @@ suspend fun INostrClient.publishAndCollectResults(
Log.d("publishAndConfirm") { "Finished with ${receivedResults.size} results" }
// Relays that never answered are still part of the verdict.
val silent = relayList - receivedResults.keys
silent.forEach { receivedResults[it] = PublishResult(false, "no response within timeout") }
return receivedResults
// Pure construction of the promised invariant: the result covers the
// full relayList, with never-answered relays reported as NO_RESPONSE.
return relayList.associateWith { receivedResults[it] ?: PublishResult(false, PublishResult.NO_RESPONSE) }
}
private class DetailedResult(
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip65RelayList/AdvertisedRelayListMutations.kt
+10 -1
View File
@@ -73,7 +73,16 @@ fun List<AdvertisedRelayInfo>.applyFacet(
val flags = LinkedHashMap<String, RW>()
for (i in this) {
urls[i.relayUrl.url] = i.relayUrl
flags[i.relayUrl.url] = i.type.rw()
val new = i.type.rw()
val old = flags[i.relayUrl.url]
flags[i.relayUrl.url] =
if (old == null) {
new
} else {
// A list may legally carry two `r` tags for the same URL (one
// `read`, one `write`) — merge the facets instead of last-wins.
RW(read = old.read || new.read, write = old.write || new.write)
}
}
val cur = flags[url.url] ?: RW(read = false, write = false)
val next = if (facet == AdvertisedRelayFacet.WRITE) cur.copy(write = present) else cur.copy(read = present)
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/FetchAllIdleTimeoutTest.kt
+31
View File
@@ -140,6 +140,37 @@ class FetchAllIdleTimeoutTest {
assertEquals(10, events.size, "fetchAll shares the idle-window semantics")
}
@Test
fun wallClockCapStopsAnEndlessTrickle() =
runTest {
val client = ScriptedClient()
var stalledRelays: Set<NormalizedRelayUrl>? = null
val feeder =
launch {
// A relay that trickles forever, always inside the idle
// window, never reaching a terminal state — without the cap
// this fetch would never return.
var i = 0
while (true) {
delay(200)
client.listener!!.onEvent(event(i++), false, relay, null)
}
}
val start = currentTime
val collected =
client.fetchAllWithHooks(
filters = mapOf(relay to listOf(Filter(kinds = listOf(1)))),
timeoutMs = 300,
maxTotalMs = 1_000,
onTimeout = { stalled, _, _ -> stalledRelays = stalled },
) { _, _ -> true }
feeder.cancel()
val elapsed = currentTime - start
assertTrue(elapsed in 1_000..1_300, "the cap must stop the fetch near maxTotalMs, took $elapsed")
assertTrue(collected.size in 4..6, "events before the cap are kept, got ${collected.size}")
assertEquals(setOf(relay), stalledRelays, "the capped relay is reported to onTimeout")
}
@Test
fun eoseStillEndsImmediately() =
runTest {
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip65RelayList/AdvertisedRelayListMutationsTest.kt
+24
View File
@@ -82,6 +82,30 @@ class AdvertisedRelayListMutationsTest {
)
}
@Test
fun splitReadWriteEntriesForSameUrlMergeToBoth() {
// Legal per NIP-65 and produced by split-marker clients: two `r` tags
// for the same URL, one `read` and one `write`. The facets must merge
// to BOTH instead of the last entry winning.
val before =
listOf(
AdvertisedRelayInfo(relay1, AdvertisedRelayType.READ),
AdvertisedRelayInfo(relay1, AdvertisedRelayType.WRITE),
)
val afterAdd = before.addFacet(relay2, AdvertisedRelayFacet.READ)
assertEquals(
listOf(relay1.url to AdvertisedRelayType.BOTH, relay2.url to AdvertisedRelayType.READ),
afterAdd.keys(),
)
val afterRemove = before.removeFacet(relay1, AdvertisedRelayFacet.WRITE)
assertEquals(listOf(relay1.url to AdvertisedRelayType.READ), afterRemove.keys())
val afterSet = before.setFacet(listOf(relay1), AdvertisedRelayFacet.WRITE)
assertEquals(listOf(relay1.url to AdvertisedRelayType.BOTH), afterSet.keys())
}
@Test
fun setFacetDemotesUnlistedAndAddsTargets() {
val before =
quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/TcpProber.kt
+16 -3
View File
@@ -27,7 +27,9 @@ import kotlinx.coroutines.withContext
import java.net.InetSocketAddress
import java.net.Socket
import java.net.URI
import java.util.concurrent.Executors
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit
/**
* Raw-socket reachability pre-probe: answers "will this relay's port even accept a
@@ -43,9 +45,20 @@ object TcpProber {
// far past the connect timeout. On the shared Dispatchers.IO those hanging lookups
// starve the crawl's own IO — measured +462s on the finishing drain at hop-3. Run
// them on a dedicated, isolated daemon pool instead so the crawl's IO is untouched.
// allowCoreThreadTimeOut: a plain fixed pool's core threads never die, and
// this object now lives in quartz — a long-lived app/relay process that
// probes once would otherwise carry 128 parked threads for its whole
// lifetime. With the 60s keep-alive the pool drains back to zero threads
// a minute after a crawl finishes.
private val probeDispatcher =
Executors
.newFixedThreadPool(128) { r -> Thread(r, "relay-probe").apply { isDaemon = true } }
ThreadPoolExecutor(
128,
128,
60L,
TimeUnit.SECONDS,
LinkedBlockingQueue(),
) { r -> Thread(r, "relay-probe").apply { isDaemon = true } }
.apply { allowCoreThreadTimeOut(true) }
.asCoroutineDispatcher()
/**